如何实现多COM口监听与电机测试任务并行处理的异步程序

需开发一个异步程序，监听三个COM口并并行处理数据。具体场景为：设备发送电机ID后，在对应COM口执行测试；当前测试完成后切换至另一项测试，同时新的电机ID会进入并执行第一项测试。已尝试使用asyncio、aioserial进行开发，以下为初始实现代码及临时解决方案，请问该如何正确处理此类场景？

初始实现代码 #

async def read_motor_id(self, motor_serial=1):
    motor_id = None
    while motor_id == None:
        # id_data = await motor_serial.read_until_async(b'\n')
        # motor_id = id_data.decode().strip()
        motor_id = input("Digite o ID do motor: ")
        if motor_id:
            print(f"Motor ID: {motor_id}")
            await self.queue.put(motor_id)
# I had to put the class cause stack couldn't let me put at the top sorry
Class MotorTest()
def __init__(self):
    self.tasks = []
    self.queue = asyncio.Queue()

async def process_queue(self, oil_serial = 1, water_serial = 1):
    while True:    
        motor_id = await self.queue.get()
        await self.process_motor_id(motor_id, oil_serial, water_serial)
        

async def process_motor_id(self, motor_id, oil_serial = 1, water_serial = 1):
    motor_type = motor_id[-2:]  # OL ou AG
    if motor_type == "OL":
        self.tasks.append(motor_id[:-2])
        await self.test_oil(motor_id, oil_serial)
    elif motor_type == "AG":
        if motor_id[:-2] in self.tasks:
            await self.test_water(motor_id, water_serial)
            
            self.tasks.remove(motor_id[:-2])
        else:
            print(f"Tarefa para motor {motor_id} não encontrada")
    await self.read_motor_id()
    
async def test_oil(self, motor_id, oil_serial =1):
    print(f"Iniciando teste de óleo para motor {motor_id}")
    await asyncio.sleep(2)  # Simulação do teste
    print(f"Finalizando teste de óleo para motor {motor_id}")
    # await oil_serial.read_async_until(aioserial.LF)

async def test_water(self, motor_id, water_serial =1):
    print(f"Iniciando teste de água para motor {motor_id}")
    await asyncio.sleep(2)  # Simulação do teste
    print(f"Finalizando teste de água para motor {motor_id}")
    # await water_serial.read_async_until(aioserial.LF)

async def main()
# motor_serial = aioserial.AioSerial(port='COM1', baudrate=9600)
# oil_serial = aioserial.AioSerial(port='COM2', baudrate=9600)
# water_serial = aioserial.AioSerial(port='COM3', baudrate=9600)

motor_test = MotorTest()

async with asyncio.TaskGroup() as tg:
    tg.create_task(motor_test.read_motor_id())
    tg.create_task(motor_test.process_queue())



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

临时解决方案代码 #

class MotorTest:
def __init__(self):
    self.tasks = []
    self.queue = asyncio.Queue()

async def read_motor_id(self, motor_serial=1):
    motor_id = None
    while motor_id == None:
        # id_data = await motor_serial.read_until_async(b'\n')
        # motor_id = id_data.decode().strip()
        motor_id = input("Digite o ID do motor: ")
        if motor_id:
            print(f"Motor ID: {motor_id}")
            await self.queue.put(motor_id)

async def process_queue(self, oil_serial = 1, water_serial = 1):
    while True:    
        motor_id = await self.queue.get()
        asyncio.create_task(self.process_motor_id(motor_id, oil_serial, water_serial))

        
async def process_motor_id(self, motor_id, oil_serial = 1, water_serial = 1):
    motor_type = motor_id[-2:]  # OL ou AG
    motor_clean = motor_id[:-2]
    if motor_type == "OL":
            if  motor_clean not in self.tasks : self.tasks.append(motor_clean)
            asyncio.create_task(self.test_oil(motor_id, oil_serial))
    elif motor_type == "AG":
            if motor_id[:-2] in self.tasks:
                asyncio.create_task(self.test_water(motor_id, water_serial))
            else:
                print(f"Tarefa para motor {motor_id} não encontrada")
                
    asyncio.run_coroutine_threadsafe(self.read_motor_id(), asyncio.get_event_loop())
async def test_oil(self, motor_id, oil_serial =1):
    print(f"Iniciando teste de óleo para motor {motor_id}")
    await asyncio.sleep(5)  # Simulação do teste
    print(f"Finalizando teste de óleo para motor {motor_id}")
    # await oil_serial.read_async_until(aioserial.LF)
async def test_water(self, motor_id, water_serial =1):
    print(f"Iniciando teste de água para motor {motor_id}")
    await asyncio.sleep(5)  # Simulação do teste
    print(f"Finalizando teste de água para motor {motor_id}")
    # await water_serial.read_async_until(aioserial.LF)

async def main():
    # motor_serial = aioserial.AioSerial(port='COM1', baudrate=9600)
    # oil_serial = aioserial.AioSerial(port='COM2', baudrate=9600)
    # water_serial = aioserial.AioSerial(port='COM3', baudrate=9600)
    motor_test = MotorTest() 
    async with  asyncio.TaskGroup() as tg:
        tg.create_task(motor_test.read_motor_id())
        tg.create_task(motor_test.process_queue())

asyncio.run(main())

现有代码的核心问题 #

• 初始代码串行阻塞：process_queue中用await self.process_motor_id处理队列，导致任务串行执行，无法并行接收新电机ID和启动测试；且process_motor_id末尾重复调用read_motor_id完全冗余。
• 临时方案操作不合理：用asyncio.run_coroutine_threadsafe重复启动read_motor_id，会导致多个读取任务同时运行，引发输入混乱和队列重复添加；self.tasks列表无线程安全保障，异步环境下易出现竞态条件。
• 测试状态管理缺失：临时方案启动测试任务后不跟踪状态，无法保证OL测试完成后再执行AG测试，也无法处理测试失败场景。

优化后实现思路 #

• 保持read_motor_id为独立异步任务，持续监听COM口并将电机ID放入队列。
• process_queue从队列取ID后，根据类型启动对应测试任务；用线程安全字典管理电机测试状态，替代原tasks列表，记录OL测试的运行/完成状态。
• 测试任务完成后更新状态，确保AG测试仅在对应OL测试完成后执行。
• 用asyncio.TaskGroup统一管理任务，避免冗余的任务启动操作。

优化后代码 #

import asyncio
from aioserial import AioSerial

class MotorTest:
    def __init__(self):
        # 字典管理电机测试状态：key为电机ID前缀，value为OL测试任务对象或完成标记
        self.motor_test_states = {}
        self.queue = asyncio.Queue()

    async def read_motor_id(self, motor_serial):
        while True:
            # 真实场景读取COM口数据
            id_data = await motor_serial.read_until_async(b'\n')
            motor_id = id_data.decode().strip()
            # 测试用输入替代（注释上方代码，启用下方代码）
            # motor_id = input("Digite o ID do motor: ")
            if motor_id:
                print(f"Motor ID recebido: {motor_id}")
                await self.queue.put(motor_id)

    async def process_queue(self, oil_serial, water_serial):
        while True:
            motor_id = await self.queue.get()
            motor_prefix = motor_id[:-2]
            motor_type = motor_id[-2:]

            if motor_type == "OL":
                # 避免重复启动同电机的OL测试
                if motor_prefix not in self.motor_test_states:
                    task = asyncio.create_task(self.test_oil(motor_id, oil_serial))
                    self.motor_test_states[motor_prefix] = task
                    # 等待OL测试完成后标记状态
                    await task
                    self.motor_test_states[motor_prefix] = "completed"
            elif motor_type == "AG":
                state = self.motor_test_states.get(motor_prefix)
                if state == "completed":
                    asyncio.create_task(self.test_water(motor_id, water_serial))
                    del self.motor_test_states[motor_prefix]
                elif isinstance(state, asyncio.Task):
                    print(f"Aguardando teste OL de {motor_prefix} para iniciar AG")
                    await state
                    asyncio.create_task(self.test_water(motor_id, water_serial))
                    del self.motor_test_states[motor_prefix]
                else:
                    print(f"Teste OL não encontrado para motor {motor_id}")
            
            self.queue.task_done()

    async def test_oil(self, motor_id, oil_serial):
        print(f"Iniciando teste de óleo para motor {motor_id}")
        # 真实场景等待COM口响应
        # await oil_serial.read_until_async(b'\n')
        await asyncio.sleep(2)
        print(f"Finalizando teste de óleo para motor {motor_id}")

    async def test_water(self, motor_id, water_serial):
        print(f"Iniciando teste de água para motor {motor_id}")
        # 真实场景等待COM口响应
        # await water_serial.read_until_async(b'\n')
        await asyncio.sleep(2)
        print(f"Finalizando teste de água para motor {motor_id}")

async def main():
    # 初始化三个异步串口
    motor_serial = AioSerial(port='COM1', baudrate=9600)
    oil_serial = AioSerial(port='COM2', baudrate=9600)
    water_serial = AioSerial(port='COM3', baudrate=9600)

    motor_test = MotorTest()

    async with asyncio.TaskGroup() as tg:
        tg.create_task(motor_test.read_motor_id(motor_serial))
        tg.create_task(motor_test.process_queue(oil_serial, water_serial))

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

关键优化点 #

• 状态管理更可靠：用字典替代列表跟踪测试状态，既避免重复启动测试，又能清晰区分测试的运行/完成状态，异步单线程环境下无需额外锁。
• 并行性与顺序性兼顾：process_queue处理队列时不阻塞其他任务，AG测试会根据OL测试状态选择立即启动或等待，既保证测试顺序，又不影响整体并行效率。
• 移除冗余操作：保持单个read_motor_id任务持续运行，避免重复启动导致的输入混乱。
• 适配真实场景：保留aioserial相关代码注释，方便快速切换到真实COM口读取逻辑。

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