Asyncio Streams执行writer.close()后仍自动重启读取数据的问题

问题原因 #

你的代码出现无限重复读取流程的核心原因有两个：

1. 服务器持续监听新连接：asyncio.start_server创建的服务器会一直监听指定端口，server.serve_forever()会让服务器永久运行，不会自动停止。即使你关闭了当前客户端连接，只要传感器设备在断开后自动发起重连（多数物联网设备默认会做这个操作），服务器就会再次调用handle_server协程，看起来像是“重启读取流程”。
2. 未终止服务器主循环：你的handle_server仅关闭了当前客户端的连接，但没有通知服务器停止监听。服务器会一直等待新的连接请求，只要传感器重连，就会重复执行读取逻辑。

解决方案 #

根据你“1秒数据传输完成后结束整个程序”的需求，需要在处理完单次连接后，主动关闭服务器并终止程序。以下是修改后的代码：

import datetime
import asyncio
import socket

async def connect_to_sq():
    _ip = "0.0.0.0"
    _port = 45454

    # 创建事件对象，用于触发服务器关闭
    shutdown_trigger = asyncio.Event()
    
    # 将关闭事件传递给处理函数
    server = await asyncio.start_server(
        lambda reader, writer: handle_server(reader, writer, shutdown_trigger),
        _ip, _port,
        family=socket.AF_INET, reuse_address=True
    )
    
    async with server:
        # 等待关闭信号，替代永久阻塞的serve_forever()
        await shutdown_trigger.wait()
        # 主动关闭服务器并等待资源释放
        server.close()
        await server.wait_closed()

async def handle_server(reader, writer, shutdown_trigger):
    print("start reading")
    init_time = datetime.datetime.now()
    end_time = init_time + datetime.timedelta(seconds=1)
    
    try:
        while datetime.datetime.now() < end_time:
            # 给读取操作添加超时，避免客户端断连后永久阻塞
            raw_data_stream = await asyncio.wait_for(
                reader.readexactly(200),
                timeout=0.5  # 可根据实际情况调整超时时间
            )
            print("time:", datetime.datetime.now()-init_time)
    except asyncio.IncompleteReadError:
        print("客户端提前断开连接")
    except asyncio.TimeoutError:
        print("读取超时，未获取到足够数据")
    finally:
        # 确保连接被正确关闭
        writer.close()
        await writer.wait_closed()
        print("Connection closed")
        # 触发服务器关闭信号
        shutdown_trigger.set()

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

关键修改点说明 #

• 添加关闭事件：用asyncio.Event作为信号，让handle_server在完成1秒读取后通知主协程关闭服务器。
• 替换serve_forever()：主协程不再永久阻塞，而是等待关闭事件触发后主动关闭服务器。
• 添加读取超时：用asyncio.wait_for包裹readexactly，避免客户端提前断开或未发送足够数据时程序永久阻塞。

如果你的需求是保持服务器运行，但每个连接仅处理1秒后关闭（不终止整个程序），只需去掉关闭服务器的逻辑，保留原有的server.serve_forever()即可——此时重复执行是正常的，因为服务器在处理完一个连接后会继续等待新的连接。

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