如何用Python asyncio/gevent实现Go select多队列等待逻辑

你这段Go代码的核心是用select多路复用监听两个通道，哪个通道有数据就处理哪个，循环持续等待。在Python里我们可以用队列替代Go的channel，分别用asyncio（原生异步）和gevent（协程）来实现相同逻辑，我给你详细写出来：

一、用asyncio实现（原生异步，Python3.7+） #

asyncio是Python官方的异步IO库，用asyncio.Queue来模拟Go的channel，再通过asyncio.wait实现类似select的多路等待：

import asyncio

async def main():
    # 初始化两个异步队列，对应Go里的chan string和chan bool
    messages = asyncio.Queue()
    signals = asyncio.Queue()

    # 模拟外部往队列里推送数据的协程（实际场景可以是网络请求、消息队列输入等）
    async def feed_queues():
        await asyncio.sleep(2)
        await messages.put("hello from message queue")
        await asyncio.sleep(1)
        await signals.put(True)
        await asyncio.sleep(3)
        await messages.put("another message")

    # 启动喂数据的后台协程
    asyncio.create_task(feed_queues())

    while True:
        # 监听两个队列的get操作，只要有一个完成就返回
        done, pending = await asyncio.wait(
            [messages.get(), signals.get()],
            return_when=asyncio.FIRST_COMPLETED
        )
        # 处理完成的任务结果
        for task in done:
            result = task.result()
            if isinstance(result, str):
                print(f"received message {result}")
            else:
                print(f"received signal {result}")
        # 取消未完成的任务，避免下次循环重复监听
        for task in pending:
            task.cancel()

# 运行主协程
asyncio.run(main())

关键说明： #

• asyncio.Queue的get()是异步方法，会挂起协程直到队列有元素
• asyncio.wait的FIRST_COMPLETED参数让我们只等待第一个完成的任务，完美对应Go select的“先到先处理”逻辑
• 每次循环后要取消未完成的任务，因为这些任务已经没用了，下次循环会重新创建新的get任务

二、用gevent实现（协程框架） #

gevent是基于Greenlet的协程框架，用gevent.queue.Queue模拟channel，通过gevent.wait实现多路等待：

import gevent
from gevent.queue import Queue

def main():
    # 初始化两个协程队列
    messages = Queue()
    signals = Queue()

    # 模拟外部往队列塞数据的绿色线程
    def feed_queues():
        gevent.sleep(2)
        messages.put("hello from message queue")
        gevent.sleep(1)
        signals.put(True)
        gevent.sleep(3)
        messages.put("another message")

    # 启动喂数据的绿色线程
    gevent.spawn(feed_queues)

    while True:
        # 把队列的get操作包装成绿色线程
        msg_task = gevent.spawn(messages.get)
        sig_task = gevent.spawn(signals.get)
        # 等待任意一个绿色线程完成（count=1表示只等第一个完成的）
        done, pending = gevent.wait([msg_task, sig_task], count=1)
        # 处理结果
        for task in done:
            result = task.value
            if isinstance(result, str):
                print(f"received message {result}")
            else:
                print(f"received signal {result}")
        # 杀死未完成的绿色线程，避免资源泄漏
        for task in pending:
            task.kill()

if __name__ == "__main__":
    main()

关键说明： #

• gevent.queue.Queue的get()是阻塞方法，但在gevent的协程模型下会自动切换上下文，不会阻塞整个进程
• gevent.wait的count=1参数让我们获取第一个完成的任务，实现类似select的多路监听
• 每次循环后杀死未完成的绿色线程，确保下次循环重新监听新的get操作

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