如何在asyncio.Protocol的connection_made中执行异步检查且不阻塞连接？

核心问题分析 #

• asyncio.Protocol的connection_made是同步回调，直接执行阻塞操作会卡住事件循环，导致无法接受新连接
• 直接用asyncio.create_task（或旧版ensure_future）启动异步检查会让connection_made提前返回，此时客户端可能在验证完成前发送数据

推荐方案：改用asyncio.start_server（基于协程的API） #

asyncio.start_server是更现代的异步服务器API，支持在连接建立时直接使用协程处理，完美匹配你的需求：

import asyncio

async def check(peername):
    await asyncio.sleep(10)  # 模拟异步IO检查
    return False  # 或True

async def handle_connection(reader, writer):
    peername = writer.get_extra_info('peername')
    print('Connection from {}'.format(peername))
    
    # 异步执行检查，不阻塞事件循环
    if not await check(peername):
        writer.close()
        await writer.wait_closed()
        return
    
    # 验证通过后处理数据
    data = await reader.read(100)
    message = data.decode()
    print('Data received: {!r}'.format(message))
    
    print('Send: {!r}'.format(message))
    writer.write(data)
    await writer.drain()
    
    print('Close the client socket')
    writer.close()
    await writer.wait_closed()

async def main():
    server = await asyncio.start_server(handle_connection, '127.0.0.1', 8888)
    
    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')
    
    async with server:
        await server.serve_forever()

asyncio.run(main())

这个方案的优势：

• 连接处理全程是协程，自然支持await异步检查，不会阻塞事件循环
• 检查完成前不会进入数据处理逻辑，避免未验证时接收数据

兼容原有Protocol的方案 #

如果必须保留asyncio.Protocol的代码结构，可以通过以下方式处理：

1. 在connection_made中启动异步检查任务
2. 暂时标记连接未验证，在data_received中先检查验证状态，未通过则直接关闭
3. 或者在启动检查前禁用transport的接收，验证通过后再启用

基础兼容版本 #

import asyncio

async def check(peername):
    await asyncio.sleep(10)
    return False

class EchoServerProtocol(asyncio.Protocol):
    def __init__(self):
        self.is_verified = False
        self.transport = None

    def connection_made(self, transport):
        self.transport = transport
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        
        # 启动异步检查任务，不阻塞connection_made返回
        asyncio.create_task(self._verify_connection(peername))

    async def _verify_connection(self, peername):
        try:
            if not await check(peername):
                self.transport.abort()
                return
            self.is_verified = True
            print(f'Connection {peername} verified')
        except Exception:
            self.transport.abort()

    def data_received(self, data):
        # 未验证前拒绝处理数据
        if not self.is_verified:
            self.transport.abort()
            return
        
        message = data.decode()
        print('Data received: {!r}'.format(message))
        
        print('Send: {!r}'.format(message))
        self.transport.write(data)
        
        print('Close the client socket')
        self.transport.close()

优化版本：暂停数据接收 #

如果想在验证完成前完全不接收客户端数据，可以在connection_made中先暂停接收：

def connection_made(self, transport):
    self.transport = transport
    peername = transport.get_extra_info('peername')
    print('Connection from {}'.format(peername))
    
    # 暂停接收数据，直到验证完成
    self.transport.pause_reading()
    asyncio.create_task(self._verify_connection(peername))

async def _verify_connection(self, peername):
    try:
        if not await check(peername):
            self.transport.abort()
            return
        self.is_verified = True
        # 验证通过后恢复接收
        self.transport.resume_reading()
        print(f'Connection {peername} verified')
    except Exception:
        self.transport.abort()

关于Protocol设计的说明 #

asyncio.Protocol的回调设计为同步并非缺陷，它是asyncio早期基于回调模型的API，适合简单的IO场景。对于需要异步处理连接建立逻辑的场景，官方更推荐使用asyncio.start_server这类基于协程的现代API。

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