django 채널을 사용하여 비동기 소켓 상태를 구현하는 방법

7385 단어 djangodjangochannelssocketpython
django를 사용하여 분석 프로젝트를 개발하는 동안 사용자가 분석을 시작할 때 상태를 표시해야 하는 문제를 발견했습니다. 우리는 django 채널을 사용하여 이것을 구현하기로 결정했습니다. 인터넷에 django 채널에 대한 많은 자습서가 있지만 모두 조금 복잡하거나 django 채널을 사용하여 채팅 응용 프로그램을 빌드합니다. 그래서 django 채널을 사용하여 사용자에게 비동기 상태를 보내는 방법을 보여주는 이 기사를 작성하기로 결정했습니다.

최소한 django 프레임워크의 기본과 django가 어떻게 작동하는지 알고 있다고 가정하고 있으므로 django 프로젝트와 django 앱을 만드는 방법을 보여주지는 않겠습니다. django 채널 구현부터 직접 시작하겠습니다.

먼저 프로젝트 channelproj를 시작하고 앱 알리미를 만듭니다.

pip를 사용하여 채널과 채널-레디스를 설치해야 합니다.

pip install channels channels-redis


redis-server를 시스템에 설치해야 합니다.

sudo apt update
sudo apt install redis-server


redis-server 상태 확인:

sudo systemctl status redis




여기에서 redis 서버가 활성 상태이고 포트 6379에서 실행 중인 것을 볼 수 있습니다.

그런 다음 django channelproj/settings.py 파일을 열고 INSTALLED_APPS에 채널과 알리미를 추가합니다.

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'channels', # Here we added channels 
    'notifier' # we will use django channels in this app
]


또한 settings.py 파일에 다음 행을 추가해야 합니다.

ASGI_APPLICATION = 'channelproj.asgi.application'

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}


채널 레이어에 대해 자세히 알아보려면 클릭하세요here.

이제 우리는 파일 notifier/consumers.py를 생성하고 다음 클래스를 추가합니다.

from channels.generic.websocket import AsyncJsonWebsocketConsumer

class StatusConsumer(AsyncJsonWebsocketConsumer):
    room_group_name = 'notify'

    async def connect(self):
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )
        await self.accept()

    async def disconnect(self, code):
        await self.channel_layer.group_discard(
            self.room_group_name, 
            self.channel_layer
        )

    async def status_notifier(self, event):
        await self.send_json(event)


여기서 우리는 AsyncJsonWebsocketConsumer를 사용하고 notify라는 그룹을 만들고 있으며 소켓 URL을 통해 모든 이벤트를 json 형식으로 보내는 status_notifier라는 함수를 정의했습니다.

다른 파일 notifier/routing.py를 만들고 소켓 URL을 추가합니다.

from django.urls import path

from .consumers import StatusConsumer

ws_urlpatterns = [
    path('ws/status/', StatusConsumer.as_asgi())
]


이제 channelproj/asgi.py 파일을 열고 아래와 같이 파일을 편집합니다.

from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application

from notifier.routing import ws_urlpatterns

os.environ.setdefault('DJANGO_SETTINGS_MODULE',
'channelproj.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": URLRouter(
        ws_urlpatterns
    )
})


여기서는 notifier/routing.py에서 ws_urlpatterns를 가져오고 ProtocolTypeRouter를 추가하고 URLRouter를 사용하여 websocket URL을 정의했습니다.

이제 알리미 앱에서 템플릿 notifier/templates/home.html을 생성하여 입력 필드가 있는 양식을 추가합니다.

  {% load static %}
  <html>
    <head>
      <title>Django Channels Status</title>
    </head>
    <body>
      <h1>Check Status</h1>

    <form action="{% url 'staus' %}" method="post">
      {% csrf_token %}
      <input type="text" placeholder="enter anything" name="TB_sample" id="TB_sample"/><br>
      <input type="submit" value="submit">
    </form>

  </body>
  </html>


이제 우리는 notifier/views.py에 두 가지 기능을 추가할 것입니다.

from django.shortcuts import render, redirect
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
import time


def index(request):
    return render(request, 'home.html')

def status_form(request):
    if request.method =='POST':
        num = int(request.POST['TB_sample'])
        progress = 10
        for i in range(num):
            room_group_name = f'notify'
            channel_layer = get_channel_layer()
            async_to_sync(channel_layer.group_send)(
                room_group_name, {
                    "type": "status.notifier",
                    "data": progress
                }
            )
            message = "Status Running"
            progress += 10
            time.sleep(1)

    context = {'message': message}
    return render(request, 'home.html', context)


여기서 우리는 post 요청이 발생할 때 호출될 status_form 함수를 정의했습니다. 이 함수는 num 변수에 post 값을 가져오고 값 10을 가진 progress라는 변수를 선언했습니다. 우리는 루프를 정의하여 실행하고 매회 10을 추가했습니다. 타임 루프가 실행되어 진행되고 1초 동안 휴면합니다. channel_layer = get_channel_layer() 라인을 추가하여 채널 레이어를 가져오고 async_to_sync 메서드를 추가하여 room_group_name에 비동기적으로 데이터를 보냅니다.

참고: 여기에서 "type": "status.notifier"를 선언함으로써 우리는 notifier/consumers.py 파일에 작성한 status_notifier 함수를 호출합니다.

또한 다음과 같이 channelproj/urls.py 파일을 편집하십시오.

from notifier.views import index,status_form

urlpatterns = [
    path('', index),
    path('status/', status_form, name="status"),
    path('admin/', admin.site.urls),
]


이제 python manage.py runserver를 사용하여 프로젝트를 실행하고 브라우저에서 http://127.0.0.1:8000/ URL을 열면 다음과 같이 표시됩니다.



소켓 상태를 확인하기 위해 Simple WebSocket Client라는 크롬 확장을 사용합니다. 확장 프로그램을 추가하고 크롬에서 확장 프로그램 아이콘을 클릭한 후 웹 소켓 URLws://127.0.0.1:8000/ws/status/을 삽입하고 열기를 클릭하면 연결이 설정되어야 합니다.

이제 프로젝트 탭으로 이동하여 입력 필드에 10을 삽입하고 양식을 제출하고 Simple WebSocket Client 탭으로 이동하십시오. 루프가 완료될 때까지 소켓에 보내는 상태가 매초 나타나는 것을 볼 수 있습니다.



얻을 수 있는 샘플 프로젝트를 github에 업로드했습니다here .

읽어 주셔서 감사합니다!

c# 중앙값 전송 및 참조 전송
stylent+prettier+VScode에서 매우 완벽한 CSS를 묘사하였다

좋은 웹페이지 즐겨찾기

개발자 우수 사이트 수집

개발자가 알아야 할 필수 사이트 100선 추천 우리는 당신을 위해 100개의 자주 사용하는 개발자 학습 사이트를 정리했습니다
best100-homepage