Django Channels页面重载后数据重复问题排查求助

问题现象 #

启动外部数据流后，用户每刷新一次页面，前端接收到的数据量就会翻倍。

问题根源 #

前端侧问题 #

1. 重复创建WebSocket连接：页面重载时localStorage的websocketConnected标记未正确重置，且未检查已有连接状态，每次刷新都会新建WebSocket连接，旧连接因未主动关闭仍保持活跃，多个连接同时接收推送数据。
2. 未处理页面卸载事件：页面刷新或关闭时未主动关闭WebSocket连接，旧连接在后台持续存在，继续接收组内消息推送。

后端侧问题 #

1. 断开连接时组成员移除逻辑错误：disconnect方法错误使用group_discard（Django Channels无此方法），正确方法应为group_remove，导致旧连接的channel_name仍保留在用户组中，每次推送都会发给所有历史连接。
2. 消息接收与任务创建逻辑错误：receive方法中同时使用await和asyncio.ensure_future，导致同步阻塞且可能重复创建任务；另外text_data是字符串而非字典，解析逻辑错误。
3. 无重复流启动校验：start_external_stream方法未检查用户是否已有活跃流，多次触发会开启多个并行流，导致数据重复推送。
4. 连接状态管理失效：active_connections是实例变量，每个连接独立维护，无法全局跟踪用户的活跃流状态。

修复方案 #

前端修复 #

1. 页面卸载时主动关闭连接：

window.addEventListener('beforeunload', () => {
  if (socket) {
    socket.close(1000, 'Page unload');
    localStorage.setItem('websocketConnected', 'false');
  }
});

2. 修正连接创建逻辑，避免重复连接：

let socket;
const storedConnected = localStorage.getItem('websocketConnected');

// 仅当无活跃连接时创建新连接
if (!socket || socket.readyState !== WebSocket.OPEN) {
  socket = new WebSocket('ws://0.0.0.0:8000/ws/binance_consumer/');
  localStorage.setItem('websocketConnected', 'true');
}

// 连接关闭时更新状态标记
socket.addEventListener("close", () => {
  localStorage.setItem('websocketConnected', 'false');
  socket = null;
});

后端修复 #

1. 修正断开连接时的组移除逻辑：

async def disconnect(self, close_code):
    await self.channel_layer.group_remove(
        self.room_group_name, 
        self.channel_name
    )
    logger.info(f"Disconnected from group: {self.room_group_name}")

2. 修复消息接收与任务创建逻辑：

async def receive(self, text_data):
    try:
        # text_data是字符串，直接解析JSON
        data = json.loads(text_data)
        if data:
            # 使用create_task异步执行，避免await阻塞
            asyncio.create_task(self.start_external_stream(data))
        else:
            asyncio.create_task(self.stop_external_stream(data))
    except Exception as e:
        logger.error(f"ERROR receiving websocket: {e}")

3. 全局跟踪用户流状态，避免重复启动：

class ExternalDataConsumer(AsyncWebsocketConsumer):
    # 类变量，全局跟踪所有用户的活跃流状态
    active_user_streams = {}

    async def start_external_stream(self, data):
        user_id = self.scope['user'].id
        # 检查用户是否已有活跃流
        if user_id in self.active_user_streams:
            logger.warning(f"Stream already running for user {user_id}")
            return
        
        # 标记用户流为活跃
        self.active_user_streams[user_id] = True
        try:
            # 此处编写你的外部流处理逻辑
            # 示例：循环获取数据并推送
            while user_id in self.active_user_streams:
                stream_data = await self.fetch_external_data(data)
                await self.send_notification(stream_data)
                await asyncio.sleep(1)
        finally:
            # 流停止后移除标记
            if user_id in self.active_user_streams:
                del self.active_user_streams[user_id]

    async def stop_external_stream(self, data):
        user_id = self.scope['user'].id
        # 移除用户的活跃流标记，触发流停止
        if user_id in self.active_user_streams:
            del self.active_user_streams[user_id]
            logger.info(f"Stream stopped for user {user_id}")

验证建议 #

1. 刷新页面后查看后端日志，确认旧连接的disconnect事件被触发，且成功从组中移除。
2. 启动流后，查看前端控制台，确认仅存在一个WebSocket连接，且数据接收量正常。
3. 多次刷新页面后，检查后端active_user_streams中同一用户仅存在一条记录。

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