如何在Python中启动两个CPU阻塞监听器并等待首个完成任务

核心要解决两个问题：一是Discum的gateway.run()是纯同步阻塞方法，会直接卡死asyncio事件循环，不能直接和Telethon的原生异步监听逻辑并行跑；二是要实现任意一端先收到消息就终止另一路监听的逻辑，不需要靠进程隔离，IO密集型的场景用线程执行器跑阻塞逻辑就足够，开销远低于多进程。

具体实现代码 #

import asyncio
from concurrent.futures import ThreadPoolExecutor
# 保留你原本的Telethon、discum相关导入和配置项

# 跨线程传递Discord监听结果的异步队列
dc_result_queue: asyncio.Queue = asyncio.Queue()

async def listentg():
    tgclient = TelegramClient('anon', conntg.tg_api_id, conntg.tg_api_hash)
    @tgclient.on(events.NewMessage(chats=conntg.canaltg, pattern=patternmatch))
    async def tg_event_handler(event):
        print("Telegram message listened")
        await tgclient.disconnect()
    await tgclient.start()
    await tgclient.run_until_disconnected()
    # 监听断开时返回来源标识，作为任务完成的信号
    return "Telegram"

def sync_dc_listener():
    """Discord端的阻塞监听逻辑，运行在独立线程中"""
    dcclient = discum.Client(token=conndc.tokendc, log=False)
    @dcclient.gateway.command
    def dc_event_handler(resp):
        if resp.event.message:
            print("Discord message listened")
            # 跨线程安全地把结果扔回主事件循环
            asyncio.run_coroutine_threadsafe(
                dc_result_queue.put("Discord"),
                asyncio.get_event_loop()
            )
            dcclient.gateway.close()
    dcclient.gateway.run()

async def listendc():
    """Discord监听的协程包装"""
    loop = asyncio.get_running_loop()
    # 把阻塞的同步监听逻辑扔到单线程执行器，不阻塞主事件循环
    await loop.run_in_executor(
        ThreadPoolExecutor(max_workers=1),
        sync_dc_listener
    )
    return await dc_result_queue.get()

async def main():
    # 把两路监听注册为asyncio任务
    tg_task = asyncio.create_task(listentg())
    dc_task = asyncio.create_task(listendc())

    # 等待任意一路任务完成
    done_set, pending_set = await asyncio.wait(
        [tg_task, dc_task],
        return_when=asyncio.FIRST_COMPLETED
    )

    # 取最先完成的任务结果
    first_task = done_set.pop()
    trigger_source = first_task.result()
    print(f"最先接收到的消息来源：{trigger_source}")

    # 取消剩下未完成的监听任务，清理资源
    for task in pending_set:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

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

关键注意点 #

• 没必要为跑Discum的阻塞逻辑开独立进程：双端监听都是网络IO等待的场景，没有需要绕开GIL的CPU密集计算，线程的创建、通信开销比进程低两个数量级，也不需要处理跨进程传参的序列化问题
• 跨线程传值必须用asyncio.run_coroutine_threadsafe：discum的逻辑跑在独立工作线程里，不能直接修改主事件循环的变量，通过这个方法可以把结果安全投递回主循环的队列，不会出现线程竞争问题
• 必须做取消逻辑：拿到第一个消息后，剩下的监听任务如果不取消，会一直挂在后台保持WebSocket连接，平白消耗网络和客户端配额
• Telethon本身是基于asyncio实现的异步客户端，不需要额外包装到执行器里，直接作为协程任务跑就可以，性能比塞到线程里好很多

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