asyncio实现TCP客户端任务偶发阻塞收不到消息、连接数与收消息数不一致问题

• 服务端连接生命周期管理缺失
  服务端on_connection协程写完数据执行drain后直接退出，未主动关闭StreamWriter对象。asyncio的StreamWriter在所属协程退出后可能被GC回收，部分场景会直接发送RST包中断连接，客户端还未读完数据连接就已断开，触发readexactly抛出EOFError或ConnectionResetError，后续计数逻辑不会执行。
• 客户端连接任务无异常捕获逻辑
  connection_to协程未做异常捕获，只要出现网络波动、连接被重置、数据长度不足等异常情况，协程会直接终止，counter +=1逻辑永远不会执行，导致收消息计数低于连接成功计数。
• 高并发短连接触发内核网络栈限制
  短时间内向同一IP端口发起7000个TCP连接属于高并发短连接场景，客户端会生成大量TIME_WAIT状态的连接，端口耗尽后部分连接会建立失败，就算连接建立成功，也可能因为内核TCP队列溢出导致丢包，服务端返回的数据无法到达客户端。
• 客户端任务调度逻辑不完善
  客户端main协程仅负责创建任务，未等待所有任务执行完成，也未收集任务执行结果，部分慢任务可能长时间挂起等待读数据，执行状态无法感知，最终统计的计数自然不匹配。

服务端调整 #

发完数据后主动关闭连接，确保数据正常发送完成后再断开，修改on_connection逻辑：

async def on_connection(r: asyncio.StreamReader, w: asyncio.StreamWriter):
    msg = struct.pack("HB", 3, 0)
    w.write(msg)
    await w.drain()
    # 新增：主动关闭写端，保证客户端收到完整数据和EOF标记
    w.close()
    await w.wait_closed()
    global counter
    counter += 1
    print(counter, "client")

客户端调整 #

新增异常捕获、并发限流、任务等待逻辑，避免端口耗尽和异常遗漏，修复后代码如下：

import asyncio

# 限制并发数，避免瞬间占满客户端端口
sem = asyncio.Semaphore(1000)
counter = 0
c_counter = 0

async def connection_to():
    async with sem:
        try:
            r, w = await asyncio.open_connection('192.168.3.2', 12345)
            global c_counter
            c_counter += 1
            print(c_counter, "connected")
            await r.readexactly(3)
            global counter
            counter += 1
            print(counter, "get_msg")
            # 主动关闭客户端侧连接
            w.close()
            await w.wait_closed()
        except Exception as e:
            # 打印异常信息，方便定位具体问题
            print(f"task exec error: {str(e)}")

async def main():
    tasks = []
    for i in range(7000):
        tasks.append(asyncio.create_task(connection_to()))
    # 等待所有连接任务执行完成
    await asyncio.gather(*tasks)
    # 输出最终统计结果
    print(f"总连接成功数: {c_counter}, 总收消息数: {counter}")

if __name__ == "__main__":
    asyncio.run(main())

可选内核优化（Linux环境） #

调整内核参数缓解TIME_WAIT端口占用问题：

# 开启TIME_WAIT状态连接复用
sysctl -w net.ipv4.tcp_tw_reuse=1
# 扩大本地可用端口范围
sysctl -w net.ipv4.ip_local_port_range="1024 65535"

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