Django + Channels + Celery 实时更新日志
1.准备工作：
-
系统为windows系统。技术实现：
python 3.6.8 django 2.2 celery 3.1.26 redis 2.10.6 django-celery 3.3.1 channels 2.4.0 channels-redis 2.4.2 -
settings.py配置。
- 注册APP
INSTALLED_APPS = [ ... # 注册django-celery "djcelery", # 注册 channel "channels", ]-
celery一些参数配置
-
一张图简单看一下celery
-
import djcelery # 加载djcelery djcelery.setup_loader() # 数据库调度 BROKER_TRANSPORT='redis' #指定redis # CELERYBEAT_SCHEDULER='djcelery.schedulers.DatabaseScheduler' # celey处理器，固定 CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0' # Broker配置，使用Redis作为消息中间件 消息队列 CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1' # BACKEND配置，这里使用redis 存储结果 # 指定任务路径。为api应用下的tasks.py文件 CELERY_IMPORTS = ('api.tasks') # CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案 #允许的内容类型， CELERY_ACCEPT_CONTENT=['pickle','json'] #任务的序列化方式 CELERY_TASK_SERIALIZER = 'json' #celery时区，定时任务使用 CELERY_TIMEZONE = 'Asia/Shanghai' # 每个worker最多执行100个任务被销毁，可以防止内存泄漏 CELERYD_MAX_TASKS_PER_CHILD = 100 # 有些情况下可以防止死锁 CELERYD_FORCE_EXECV = True # 设置并发的worker数量 CELERYD_CONCURRENCY = 4 # 允许重试 CELERY_ACKS_LATE = True # 单个任务的最大运行时间，超过就杀死 CELERYD_TASK_TIME_LEMIT = 12 * 30- channels一些配置，注册channels需要指定ASGI路由地址
# 指定ASGI的路由地址 # 指定api 应用下routing.py ASGI_APPLICATION = 'api.routing.application'# CHANNEL_LAYERS = { 'default': { 'BACKEND': 'channels_redis.core.RedisChannelLayer', 'CONFIG': { "hosts": [('127.0.0.1', 6379)], }, }, }- 假数据
# 这里字典形式封装路经代表日志文件路径 TAILF = { 1: r'J:\djangoSerializers\api\log_recored\sheet1', 2: r'J:\djangoSerializers\api\log_recored\sheet2', }
1.5文件目录如下：
djangoSerializers
|__ api
|___ __init__.py
|___ taksk.py # 为创建
|___ views.py
|___ ...
|__ djangoSerializers
|___ __init__.py
|___ settings.py
|___ urls.py
|___ wsgi.py
|___ celery.py #为创建
2.celery
-
在项目目录下二级目录下创建celery.py
from django.conf import setting from __future__ import absolute_import, unicode_literals from celery import Celery, platforms from django.conf import settings import os # 设置当前django环境 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "djangoSerializers.settings") # 实例化Celery对象 app = Celery("djangoSerializers") # 加载配置文件，并使用CELERY前缀 app.config_from_object("django.conf:settings", namespace='CELERY') # celery不能root用户启动解决 platforms.C_FORCE_ROOT = True # 去寻找每个app下的tasks.py文件 app.autodiscover_tasks(lambda :settings.INSTALLED_APPS) -
在该目录
__init___.py添加# 这是为了确保在django启动时启动 celery from __future__ import absolute_import from .celery import app as celery_app -
在app应用下创建
tasks.py，用于celery异步任务处理。from celery import shared_task from asgiref.sync import async_to_sync from channels.layers import get_channel_layer @shared_task def tailf(id,channel_name): # 暂时先空出来
3.channel
-
因我们之前在
settings.py指定ASGI_APPLICATION路径，于是在api下创建routing.py：from channels.auth import AuthMiddlewareStack from channels.routing import ProtocolTypeRouter, URLRouter from django.urls import re_path from tailf.consumers import TailfConsumer class TokenAuthMiddle: def __init__(self,inner): self.inner = inner def __call__(self,scope): return self.inner(scope) # 这里为了简单验证直接返回当前对象，也可以自定义或者使用内置 AuthMiddlewareStack TokenAuthMiddlewareStack = lambda inner: TokenAuthMiddle(AuthMiddlewareStack(inner)) # 指向处理websocket的类视图函数 TailfConsumer application = ProtocolTypeRouter({ "websocket": TokenAuthMiddlewareStack(URLRouter([ re_path(r'^ws/tailf/(?P<id>\d+)/$', TailfConsumer), ]) ) }) -
在当前项目目录下新建
consumers.py用于websocket的连接和断开。import json from channels.generic.websocket import WebsocketConsumer # 这里tailf是要执行异步任务 from api.tasks import tailf class TailfConsumer(WebsocketConsumer): def connect(self): # 通过获取id来执行异步任务 self.file_id = self.scope["url_route"]["kwargs"]["id"] self.result = tailf.delay(self.file_id, self.channel_name) self.accept() def disconnect(self, code): # 终止执行中task self.result.revoke(terminate=True) print("disconnect:",self.file_id,self.channel_name) def send_message(self,event): # 发送给客户端消息 self.send(text_data=json.dumps({ "message":event["message"] }))
4.其他逻辑配置
-
url.py
urlpatterns = [ ... url(r'tailf', views.tailf_view, name='tailf-url'), ] -
视图函数
def tailf_view(request): logDict = settings.TAILF return render(request,"index.html",{"logDict": logDict}) -
templates
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> <div class="col-sm-8"> <select class="form-control" id="file"> <option value="">选择要监听的日志</option> {% for k,v in logDict.items %} <option value="{{ k }}">{{ v }}</option> {% endfor %} </select> </div> <div class="col-sm-2"> <input class="btn btn-success btn-block" type="button" onclick="connect()" value="开始监听"/><br/> </div> <div class="col-sm-2"> <input class="btn btn-warning btn-block" type="button" onclick="goclose()" value="终止监听"/><br/> </div> <div class="col-sm-12"> <textarea class="form-control" id="chat-log" disabled rows="20"></textarea> </div> <script src="https://cdn.bootcss.com/jquery/3.5.0/jquery.min.js"></script> </body> <script> function connect() { if ( $('#file').val() ) { var url = 'ws://' + window.location.host + '/ws/tailf/' + $('#file').val() + '/'; window.chatSocket = new WebSocket(url); // 当浏览器接收到websocket服务器发送过来的数据时，就会触发onmessage消息，参数e包含了服务端发送过来的数据 chatSocket.onmessage = function(e) { var data = JSON.parse(e.data); var message = data['message']; document.querySelector('#chat-log').value += (message); // 跳转到页面底部 $('#chat-log').scrollTop($('#chat-log')[0].scrollHeight); }; // 如果连接失败，或者发送、接收数据失败，或者数据处理出错都会触发onerror消息 chatSocket.onerror = function(e) { console.error('服务端连接异常！') }; // 当浏览器接收到websocket服务器发送过来的关闭连接请求时，会触发onclose消息 chatSocket.onclose = function(e) { console.error('websocket已关闭！') }; } else { console.log('请选择要监听的日志文件') } }; function goclose() { // 用于关闭连接 window.chatSocket.close(); window.chatSocket.onclose = function(e) { console.log('已终止日志监听！') }; } </script> </html>- 当然还有
onopen:当浏览器和websocket服务端连接成功后会触发onopen消息。
- 当然还有
5.启动worker节点：
- 终端执行：
python3 manage.py celery worker -l INFO
- 启动项目测试一下吧。