基于Django-Channels的AsyncWebSocketConsumer实现答题倒计时与定时发题方案咨询

核心设计思路 #

• 用异步定时任务处理倒计时推送，避免阻塞WebSocket连接
• 分离倒计时逻辑与题目推送逻辑，保证模块独立性
• 维护每个连接的独立状态（当前题目、剩余时间、任务状态），避免多用户状态混乱

关键代码实现 #

1. AsyncWebSocketConsumer 核心代码 #

import asyncio
from channels.generic.websocket import AsyncWebSocketConsumer
import json

class QuizConsumer(AsyncWebSocketConsumer):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.current_question = 1  # 当前题目序号
        self.countdown_task = None  # 存储倒计时异步任务
        self.per_question_time = 30  # 每题固定时长

    async def connect(self):
        await self.accept()
        # 连接建立后直接启动第一题流程
        await self.init_question_flow()

    async def disconnect(self, close_code):
        # 断开连接时取消未完成的定时任务，防止内存泄漏
        if self.countdown_task:
            self.countdown_task.cancel()

    async def init_question_flow(self):
        # 先取消可能存在的旧任务
        if self.countdown_task:
            self.countdown_task.cancel()
        
        # 推送当前题目数据
        await self.send(text_data=json.dumps({
            'type': 'new_question',
            'question_id': self.current_question,
            'content': f'第{self.current_question}题：这里替换为实际题目内容',
            'total_time': self.per_question_time
        }))

        # 启动倒计时推送任务
        self.countdown_task = asyncio.create_task(self.push_countdown_updates())

    async def push_countdown_updates(self):
        try:
            # 从设定时长倒推至0
            for remaining in range(self.per_question_time, -1, -1):
                await self.send(text_data=json.dumps({
                    'type': 'countdown',
                    'remaining_time': remaining
                }))
                await asyncio.sleep(1)
            
            # 倒计时结束后自动切换下一题
            self.current_question += 1
            await self.init_question_flow()
        except asyncio.CancelledError:
            # 任务被取消时正常退出，避免抛出异常
            pass

    # 可选：处理客户端提交的答案
    async def receive(self, text_data):
        data = json.loads(text_data)
        user_answer = data.get('answer')
        # 此处可添加答案验证、计分等逻辑，比如：
        # await self.send(text_data=json.dumps({'type': 'answer_result', 'is_correct': True}))

2. 路由配置（routing.py） #

from django.urls import re_path
from .consumers import QuizConsumer

websocket_urlpatterns = [
    re_path(r'ws/quiz/session/$', QuizConsumer.as_asgi()),
]

3. 前端基础交互代码（JavaScript） #

const ws = new WebSocket(`ws://${window.location.host}/ws/quiz/session/`);

ws.onmessage = (event) => {
    const data = JSON.parse(event.data);
    switch(data.type) {
        case 'new_question':
            document.getElementById('question-container').innerText = data.content;
            break;
        case 'countdown':
            document.getElementById('countdown-display').innerText = `剩余时间：${data.remaining_time}秒`;
            break;
    }
};

ws.onclose = () => {
    console.log('答题连接已关闭');
};

设计亮点 #

• 异步非阻塞：用asyncio.create_task和asyncio.sleep实现倒计时，不影响WebSocket接收用户答案等其他操作
• 状态隔离：每个WebSocket连接独立维护自身的题目和任务状态，不会出现多用户互相干扰的情况
• 资源清理：断开连接时主动取消定时任务，避免无效任务占用系统资源

注意事项 #

• 如果需要支持多房间答题，可在connect方法中加入分组逻辑（如await self.channel_layer.group_add(room_id, self.channel_name)）
• 实际项目中，题目内容需从数据库异步获取，避免阻塞事件循环
• 生产环境需配置Channels的ASGI服务器（如Daphne），并使用Redis作为通道层后端（适配分布式部署场景）

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