Django Channels服务端断开问题：日志查看器刷新后连接失败

我使用Django与Channels开发日志文件查看器，首次连接可正常运行，但刷新前端后WebSocket连接失败。排查发现首次连接的循环仍在运行，WebSocket关闭时并未触发disconnect方法。

消费者代码 #

import time
from channels.exceptions import StopConsumer
from channels.generic.websocket import WebsocketConsumer
class LogConsumer(WebsocketConsumer):
    end = False
    def connect(self):
        self.accept()
        filename = self.scope['url_route']['kwargs']['filepath']
        file_fullname = "/tmp/" + filename
        print('file name: %s' % file_fullname)
        with open(file_fullname, 'r') as f:
           while True:
               time.sleep(1)
               if self.end:
                   break
               new_line = f.readline()
               print('Loop is still running ....')
               if new_line:
                   print('sending data: %s' % new_line)
                   self.send(text_data=new_line)

    def disconnect(self, event):
        print('websocket disconnected...', event)
        self.end = True
        raise StopConsumer()

Daphne 输出日志 #

192.168.1.2:63267 - - [03/Jun/2023:16:25:09] "WSCONNECTING /ws/message/" - -
192.168.1.2:63267 - - [03/Jun/2023:16:25:09] "WSCONNECT /ws/message/" - -
file name: /tmp/message
Loop is still running ....
sending data:

Loop is still running ....
sending data: 1

Loop is still running ....
sending data: 2

Loop is still running ....
sending data: 3

192.168.1.2:63267 - - [03/Jun/2023:16:25:14] "WSDISCONNECT /ws/message/" - -
192.168.1.2:63273 - - [03/Jun/2023:16:25:14] "WSCONNECTING /ws/message/" - -
Loop is still running ....
sending data: 4

Loop is still running ....
sending data: 5

问题根源 #

1. 阻塞式循环占用线程：connect方法里的while True循环搭配time.sleep(1)是同步阻塞操作，会一直占用当前线程，导致Channels无法处理后续的disconnect事件——线程被死循环卡住，根本没机会执行断开逻辑。
2. 类属性共享问题：end是类属性，所有连接实例会共享这个变量，逻辑上存在并发冲突风险。

修复方案 #

方案一：改用异步消费者（推荐）

使用AsyncWebsocketConsumer，用异步睡眠替代同步睡眠，避免阻塞事件循环，同时将end改为实例属性：

import asyncio
from channels.exceptions import StopConsumer
from channels.generic.websocket import AsyncWebsocketConsumer

class LogConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.end = False  # 实例属性，每个连接独立控制
        await self.accept()
        filename = self.scope['url_route']['kwargs']['filepath']
        file_fullname = "/tmp/" + filename
        print('file name: %s' % file_fullname)
        
        with open(file_fullname, 'r') as f:
            while not self.end:
                await asyncio.sleep(1)  # 异步睡眠，不阻塞事件循环
                new_line = f.readline()
                print('Loop is still running ....')
                if new_line:
                    print('sending data: %s' % new_line)
                    await self.send(text_data=new_line)

    async def disconnect(self, event):
        print('websocket disconnected...', event)
        self.end = True
        raise StopConsumer()

方案二：同步消费者加独立线程

如果必须使用同步消费者，将文件监听逻辑放到单独线程中，避免阻塞主线程处理断开事件：

import time
import threading
from channels.exceptions import StopConsumer
from channels.generic.websocket import WebsocketConsumer

class LogConsumer(WebsocketConsumer):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.end = False
        self.watch_thread = None

    def connect(self):
        self.accept()
        filename = self.scope['url_route']['kwargs']['filepath']
        file_fullname = "/tmp/" + filename
        print('file name: %s' % file_fullname)
        
        # 启动独立线程处理文件监听
        self.watch_thread = threading.Thread(target=self._watch_log_file, args=(file_fullname,))
        self.watch_thread.start()

    def _watch_log_file(self, file_fullname):
        with open(file_fullname, 'r') as f:
            while not self.end:
                time.sleep(1)
                new_line = f.readline()
                print('Loop is still running ....')
                if new_line:
                    print('sending data: %s' % new_line)
                    self.send(text_data=new_line)

    def disconnect(self, event):
        print('websocket disconnected...', event)
        self.end = True
        if self.watch_thread:
            self.watch_thread.join()  # 等待线程结束
        raise StopConsumer()

内容的提问来源于stack exchange，提问作者lufy