Python asyncio客户端-服务器协程问题：加入已有房间报错

错误根源 #

错误提示 TypeError: a coroutine was expected, got None 是因为 client.write() 是同步方法，不返回协程对象，但你用 asyncio.create_task() 包装了它。create_task 要求传入协程对象，传入 write() 返回的 None 就会触发该错误。

当房间里没有其他客户端时，broadcast 中的 tasks 为空，不会执行 await asyncio.gather(*tasks)，因此不会报错；只有当房间存在其他客户端时，才会执行 create_task 触发错误。

修复方案 #

修改服务器的 broadcast 方法，将写入操作封装成包含 drain() 的协程（异步IO中必须等待 drain() 确保数据发送完成），再用 create_task 执行。

关键修改代码 #

添加一个辅助协程 send_message，并修改 broadcast 方法：

async def send_message(self, client, message):
    client.write(message.encode())
    await client.drain()

async def broadcast(self, message, room, sender):
    tasks = []
    for client in self.rooms[room]:
        if client != sender:
            task = asyncio.create_task(self.send_message(client, message))
            tasks.append(task)
    if tasks:
        await asyncio.gather(*tasks)

完整修复后的服务器代码 #

import asyncio

class ChatServer:
    def __init__(self):
        self.clients = {}
        self.rooms = {}

    async def handle_client(self, reader, writer):
        addr = writer.get_extra_info('peername')
        print(f'Connected by {addr}')

        # Send the list of available rooms to the client
        available_rooms = '\n'.join(self.rooms.keys())
        print("---->" + available_rooms)
        writer.write(f'Available Rooms: {available_rooms}\n'.encode())
        await writer.drain()

        while True:
            message = await reader.readline()
            if not message:
                break
            message = message.decode().strip()
            if not message:
                continue
            command = message.split()[0]

            if command == "LIST":
                available_rooms = '\n'.join(self.rooms.keys())
                writer.write(f'Available Rooms: {available_rooms}\n'.encode())
                await writer.drain()
            elif command == "JOIN":
                room = message.split()[1]
                if room not in self.rooms:
                    writer.write(f"{room} room not found. creating it\n".encode())
                    self.rooms[room] = []
                    await writer.drain()
                # 移除旧房间记录（如果已在其他房间）
                if writer in self.clients:
                    old_room = self.clients[writer]
                    self.rooms[old_room].remove(writer)
                self.clients[writer] = room
                self.rooms[room].append(writer)
                writer.write(f'You joined room: {room}!\n'.encode())
                await writer.drain()
                await self.broadcast(f'{addr} has joined room {room}!\n', room, writer)

            elif command == "LEAVE":
                if writer not in self.clients:
                    writer.write(f'You are not currently in a room!\n'.encode())
                    await writer.drain()
                else:
                    room = self.clients[writer]
                    writer.write(f'You left room {room}!\n'.encode())
                    await writer.drain()
                    self.clients.pop(writer)
                    self.rooms[room].remove(writer)
                    await self.broadcast(f'{addr} has left room {room}!\n', room, writer)

                    # 房间为空则移除
                    if not self.rooms[room]:
                        self.rooms.pop(room)
            else:
                try:
                    await self.broadcast(f'{addr}: {message}\n', self.clients[writer], writer)
                except KeyError:
                    writer.write(f'You are not currently in a room!\n'.encode())
                    await writer.drain()
                    continue

        print(f'Disconnected {addr}')
        # 客户端断开时清理房间记录
        if writer in self.clients:
            room = self.clients[writer]
            self.clients.pop(writer)
            self.rooms[room].remove(writer)
            await self.broadcast(f'{addr} has disconnected!\n', room, writer)
            if not self.rooms[room]:
                self.rooms.pop(room)
        writer.close()
        await writer.wait_closed()

    async def send_message(self, client, message):
        client.write(message.encode())
        await client.drain()

    async def broadcast(self, message, room, sender):
        tasks = []
        for client in self.rooms[room]:
            # 排除发送者
            if client != sender:
                task = asyncio.create_task(self.send_message(client, message))
                tasks.append(task)
        if tasks:
            await asyncio.gather(*tasks)

    def start(self):
        loop = asyncio.get_event_loop()
        coro = asyncio.start_server(self.handle_client, '127.0.0.1', 5323, loop=loop)
        server = loop.run_until_complete(coro)

        print('Serving on {}'.format(server.sockets[0].getsockname()))
        try:
            loop.run_forever()
        except KeyboardInterrupt:
            pass

        server.close()
        loop.run_until_complete(server.wait_closed())
        loop.close()

server = ChatServer()
server.start()

额外优化说明 #

1. 客户端断开时自动清理房间内的无效连接记录
2. 处理 JOIN 命令时，自动移除客户端在旧房间的记录
3. 添加空消息判断，避免 split() 报错

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