django chnanels

https://www.youtube.com/watch?v=Wv5jlmJs2sU

https://youtu.be/Wv5jlmJs2sU

https://channels.readthedocs.io/en/stable/topics/consumers.html

settings.py

ASGI_APPLICATION = 'lbs_maneger.routing.application'

#内存通道层
#
# 请勿在生产中使用
#
# 内存通道层与每个进程一起作为单独的层运行。这意味着不可能进行跨进程消息传递。由于通道层的核心价值是提供分布式消息传递，因此内存使用将导致性能欠佳，并最终导致多实例环境中的数据丢失。
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels.layers.InMemoryChannelLayer"
    }
}

asgi.py

import os
import django
from django.core.asgi import get_asgi_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'lbs_maneger.settings')
django.setup()
application = get_asgi_application()

routing.py

# -*- coding: utf-8 -*-
"""
@Time    : 2021/2/2 11:26
@Author  : jon
@File    : routing.py.py
"""

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path, re_path
from lbs_maneger.consumer import ChatConsumer
from lbs_maneger.consumer import TailfConsumer


application = ProtocolTypeRouter({
    'websocket': AuthMiddlewareStack(
        URLRouter([
            path("ws/chat/", ChatConsumer.as_asgi()),
            
        ])
    )
})

consumer.py

import json
import sys
import asyncio
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.layers import get_channel_layer
from lbs_maneger.settings import logs
from lbs_maneger.settings import msg
from lbs_maneger.tail_task import tailf

class TailfConsumer(AsyncWebsocketConsumer):
    groups = ["tailf"]

    async def connect(self):
        logs.info("channel_name: {}".format(self.channel_name))
        self.file_id = 1
        self.result = tailf(self.file_id, self.channel_name)
        print(222, self.result)
        await self.accept()

    async def disconnect(self, close_code):
        # 中止执行中的receive
        self.result.revoke(terminate=True)
        logs.info("disconnect channel_name: {}".format(self.channel_name))

    async def receive(self, text_data=None, bytes_data=None):
        await self.tailf(self.channel_name)

    async def tailf(self, channel_name):
        channel_layer = get_channel_layer()
        filename = "/var/log/nginx/access.log"
        try:
            with open(filename) as f:
                f.seek(0, 2)
                while True:
                    line = f.readline()
                    if line:
                        data = json.dumps(line)
                        print(channel_name, line)
                        await self.send(text_data=data)

                    else:
                        logs.info("filename sleep ...")
                        await asyncio.sleep(1)
        except Exception as e:
            print(e)

前端测试

ws = new WebSocket('ws://192.168.30.134:8044/ws/tailf/');
ws.onmessage = event => console.log(event.data)
ws.send("tailf")

image.png