Django Channels报错Event loop is closed：倒计时功能修复咨询

需要实现倒计时展示功能并启动游戏循环，当前代码可正常发送消息，但反复触发RuntimeError: Event loop is closed错误。曾考虑拆分Consumer，但不确定能否解决问题。

报错信息 #

Task exception was never retrieved
future: <Task finished name='Task-60' coro=<Connection.disconnect() done, defined at D:\Programming\Fullstack\geogame\geogame_backend\env\lib\site-packages\redis\asyncio\connection.py:819> exception=RuntimeError('Event loop is closed')>
Traceback (most recent call last):
  File "D:\Programming\Fullstack\geogame\geogame_backend\env\lib\site-packages\redis\asyncio\connection.py", line 828, in disconnect
    self._writer.close()  # type: ignore[union-attr]
  File "C:\Program Files\Python310\lib\asyncio\streams.py", line 337, in close
    return self._transport.close()
  File "C:\Program Files\Python310\lib\asyncio\selector_events.py", line 698, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "C:\Program Files\Python310\lib\asyncio\base_events.py", line 753, in call_soon
    self._check_closed()
  File "C:\Program Files\Python310\lib\asyncio\base_events.py", line 515, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

原Consumer代码 #

class LobbyConsumer(WebsocketConsumer):

    def __init__(self, *args, **kwargs):
        super().__init__(args, kwargs)
        self.players = None
        self.game_finished = None
        self.host = None
        self.user = None

        self.lobby_code = None
        self.lobby_group_code = None
        self.lobby = None

    def connect(self):
        self.lobby_code = self.scope['url_route']['kwargs']['lobby_code']

        if Lobby.objects.filter(code=self.lobby_code).exists():
            self.lobby = Lobby.objects.get(code=self.lobby_code)
        else:
            print("DOESNT EXIST")
            self.accept()
            self.send(json.dumps({"type": "error", "message": "no_match"}))
            self.close()
            return


        self.lobby_group_code = f"lobby_{self.lobby_code}"

        self.host = Lobby.objects.select_related('host').get(code=self.lobby_code).host

        self.user = self.scope['user']

        if self.user.is_authenticated:
            self.accept()
            print(True)
        else:
            self.close(code=4004)
        self.lobby.users.add(self.user)
        self.lobby.save()

        players_queryset = self.lobby.users.all()


        self.players = []

        for player in players_queryset:
            self.players.append({
                "username": player.username,
                "email": player.email
            })

        async_to_sync(self.channel_layer.group_add)(
            self.lobby_group_code,
            self.channel_name,
        )

        async_to_sync(self.channel_layer.group_send)(
            self.lobby_group_code,
            {
                'type': 'status',
                'message': "Welcome",
                'has_started': self.lobby.has_started,
                'host': self.host.username,
                'players': self.players,
                'lobby_code': self.lobby_code,
                'max_players': self.lobby.max_players,
            }
        )

    def start_game(self):
        countdown = 4
        for i in range(countdown):
            async_to_sync(self.channel_layer.group_send)(
                self.lobby_group_code,
                {
                    'type': 'status',
                    'has_started': self.lobby.has_started,
                    'host': self.host.username,
                    'players': self.players,
                    'lobby_code': self.lobby_code,
                    'countdown': countdown - 1
                }
            )


            sleep(1)

            countdown -= 1



    def receive(self, text_data=None, bytes_data=None):
        command = ""
        text_data_json = json.loads(text_data)
        print(text_data_json)

        if "command" in text_data_json:
            command = text_data_json['command']

            if command == "start":
                self.lobby.has_started = True
                print("start")

                self.lobby.save()

                async_to_sync(self.channel_layer.group_send)(
                    self.lobby_group_code,
                    {
                        'type': 'status',
                        'message': "Welcome",
                        'has_started': self.lobby.has_started,
                        'host': self.host.username,
                        'players': self.players,
                        'lobby_code': self.lobby_code,
                        'countdown': 3
                    }
                )

                thread = Thread(target=self.start_game, args=())
                thread.start()


    def disconnect(self, close_code):
        print(f"Connection Cancelled")

        async_to_sync(
            self.channel_layer.group_discard(
                self.lobby_group_code,
                self.channel_name,
            ))

        self.close()

    def chat_message(self, event):
        self.send(text_data=json.dumps(event))

    def status(self, event):
        self.send(text_data=json.dumps(event))

1. 线程与事件循环冲突：使用Thread+time.sleep启动倒计时，新线程中调用async_to_sync会创建临时事件循环，线程结束后循环被销毁，Redis异步连接尝试关闭时触发"Event loop is closed"错误。
2. 错误的频道组方法：disconnect中使用了不存在的group_discard方法，正确方法应为group_remove。
3. 数据库查询冗余：connect方法中两次查询Lobby对象，造成不必要的数据库开销。

核心修复步骤 #

1. 切换到异步Consumer #

使用AsyncWebsocketConsumer替代WebsocketConsumer，全程用异步操作避免线程冲突：

• 用await直接调用Channels异步API，无需async_to_sync
• 用asyncio.sleep替代time.sleep，不阻塞事件循环
• 用sync_to_async包装Django ORM操作（ORM为同步实现）

2. 用异步任务替代线程 #

收到"start"命令后，用asyncio.create_task启动倒计时任务，无需手动创建线程。

3. 修复频道组退出逻辑 #

将group_discard改为正确的group_remove，确保连接断开时正确退出频道组。

优化思路 #

• 拆分Consumer：将逻辑拆分为LobbyLifecycleConsumer（处理连接、玩家加入/退出）和GameConsumer（处理倒计时、游戏循环），通过频道组通信解耦，避免单一Consumer负担过重。
• 减少数据库查询：合并重复的Lobby查询，用select_related或prefetch_related预加载关联数据。
• 统一状态管理：通过Redis或频道组维护游戏状态，避免在Consumer中直接频繁操作数据库。

import asyncio
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import sync_to_async
import json
from .models import Lobby

class LobbyConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.lobby_code = self.scope['url_route']['kwargs']['lobby_code']
        self.lobby_group_code = f"lobby_{self.lobby_code}"
        
        # 异步查询Lobby，合并关联查询
        try:
            self.lobby = await sync_to_async(Lobby.objects.select_related('host').get)(code=self.lobby_code)
        except Lobby.DoesNotExist:
            await self.accept()
            await self.send(json.dumps({"type": "error", "message": "no_match"}))
            await self.close()
            return

        self.user = self.scope['user']
        if not self.user.is_authenticated:
            await self.close(code=4004)
            return
        
        await self.accept()
        # 异步添加用户到 lobby
        await sync_to_async(self.lobby.users.add)(self.user)
        
        # 异步获取玩家列表
        players_queryset = await sync_to_async(list)(self.lobby.users.all())
        self.players = [
            {"username": p.username, "email": p.email} for p in players_queryset
        ]
        self.host = self.lobby.host

        # 加入频道组
        await self.channel_layer.group_add(
            self.lobby_group_code,
            self.channel_name
        )

        # 发送初始状态
        await self.channel_layer.group_send(
            self.lobby_group_code,
            {
                'type': 'status',
                'message': "Welcome",
                'has_started': self.lobby.has_started,
                'host': self.host.username,
                'players': self.players,
                'lobby_code': self.lobby_code,
                'max_players': self.lobby.max_players,
            }
        )

    async def start_game(self):
        countdown = 3
        while countdown >= 0:
            await self.channel_layer.group_send(
                self.lobby_group_code,
                {
                    'type': 'status',
                    'has_started': self.lobby.has_started,
                    'host': self.host.username,
                    'players': self.players,
                    'lobby_code': self.lobby_code,
                    'countdown': countdown
                }
            )
            if countdown > 0:
                await asyncio.sleep(1)
            countdown -= 1
        
        # 倒计时结束后启动游戏循环
        await self.start_game_loop()

    async def start_game_loop(self):
        # 这里实现你的游戏循环逻辑
        pass

    async def receive(self, text_data=None, bytes_data=None):
        text_data_json = json.loads(text_data)
        command = text_data_json.get('command')

        if command == "start":
            self.lobby.has_started = True
            await sync_to_async(self.lobby.save)()

            # 发送启动通知
            await self.channel_layer.group_send(
                self.lobby_group_code,
                {
                    'type': 'status',
                    'message': "Game starting...",
                    'has_started': self.lobby.has_started,
                    'host': self.host.username,
                    'players': self.players,
                    'lobby_code': self.lobby_code,
                    'countdown': 3
                }
            )

            # 用异步任务启动倒计时
            asyncio.create_task(self.start_game())

    async def disconnect(self, close_code):
        print(f"Connection Cancelled")
        # 正确退出频道组
        await self.channel_layer.group_remove(
            self.lobby_group_code,
            self.channel_name
        )
        await self.close()

    async def chat_message(self, event):
        await self.send(text_data=json.dumps(event))

    async def status(self, event):
        await self.send(text_data=json.dumps(event))

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