使用Python asyncio实现双端口同时收包的UDP服务器相关问题求助

问题出在代码逻辑的两处缺陷：

• 首先是明显的代码笔误：你两次调用receive()方法都用了元数据端口的meta_ep实例，从头到尾没有调用过语音端口spch_ep的接收方法，本来应该收语音报文的位置实际一直在等元数据端口的包，这是直接诱因
• 其次是串行接收的逻辑本身不合理：就算修正了笔误，先await元数据接收、再await语音接收的串行逻辑也会导致两类报文互相阻塞——只要其中一个端口暂时没有新包，程序就会一直卡在对应await步骤，另一个端口就算到了大量报文也无法处理，内核缓冲区满后就会出现丢包、顺序错乱的问题。

收包原理 #

asyncio的DatagramProtocol是事件驱动模型：你给两个端口分别注册Protocol实例后，事件循环会在内核通知有UDP包到达时，自动调用对应端口Protocol实例的datagram_received方法，不需要手动写循环接收逻辑，两个端口的收包、处理过程完全独立，不会出现互相阻塞的问题。

具体实现方案 #

你可以给两个端口分别定义不同的Protocol类区分报文类型，再用全局通话状态存储关联同个通话的元数据和语音数据，示例实现如下：

import asyncio
from collections import defaultdict

local_ip = '0.0.0.0'
metadata_port = 3001
speech_port = 3002

# 全局通话状态存储，key为call_id，value存储通话时间、语音片段等数据
call_sessions = defaultdict(lambda: {
    "start_time": None,
    "end_time": None,
    "speech_chunks": []
})

class MetaDataProtocol(asyncio.DatagramProtocol):
    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        msg = data.decode().strip()
        if msg.startswith("START_CALL_"):
            call_id = msg.split("_")[-1]
            call_sessions[call_id]["start_time"] = asyncio.get_event_loop().time()
            print(f"通话{call_id}已开始")
        elif msg.startswith("END_CALL_"):
            call_id = msg.split("_")[-1]
            call_sessions[call_id]["end_time"] = asyncio.get_event_loop().time()
            # 此处可扩展通话数据落盘、资源释放逻辑
            print(f"通话{call_id}已结束，共收到{len(call_sessions[call_id]['speech_chunks'])}段语音")

class SpeechProtocol(asyncio.DatagramProtocol):
    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        msg = data.decode().strip()
        if msg.startswith("SPEECH_"):
            parts = msg.split("_")
            call_id = parts[1]
            idx = parts[2]
            call_sessions[call_id]["speech_chunks"].append((idx, data))
            print(f"收到通话{call_id}的第{idx}段语音")

def main():
    loop = asyncio.get_event_loop()
    # 启动元数据端口服务
    meta_task = loop.create_datagram_endpoint(
        MetaDataProtocol,
        local_addr=(local_ip, metadata_port)
    )
    # 启动语音端口服务
    speech_task = loop.create_datagram_endpoint(
        SpeechProtocol,
        local_addr=(local_ip, speech_port)
    )
    loop.run_until_complete(meta_task)
    loop.run_until_complete(speech_task)
    loop.run_forever()

if __name__ == '__main__':
    main()

内容的提问来源于stack exchange，提问作者Jung Hyuk Lee