Python中模拟Go语言的select多路复用机制

import threading
import queue # Python 3 uses queue, Python 2 uses Queue

def main_direct_simulation():
    # 模拟Go的channel，使用Python的Queue.Queue
    c1 = queue.Queue(maxsize=0)
    c2 = queue.Queue(maxsize=0)
    quit_channel = queue.Queue(maxsize=0) # 避免与Python内置的quit冲突

    # 生产者函数1：向c1发送数据，最后向quit_channel发送退出信号
    def producer_func1():
        for i in range(10):
            c1.put(i)
        quit_channel.put(0)

    # 启动生产者线程1
    threading.Thread(target=producer_func1).start()

    # 生产者函数2：向c2发送数据
    def producer_func2():
        for i in range(2):
            c2.put(i)

    # 启动生产者线程2
    threading.Thread(target=producer_func2).start()

    # 中央消息聚合队列
    combined_queue = queue.Queue(maxsize=0)

    # 监听并转发消息的守护线程函数
    def listen_and_forward(source_queue):
        while True:
            # 从源队列获取消息，并将其与源队列本身一起放入聚合队列
            message = source_queue.get()
            combined_queue.put((source_queue, message))

    # 为每个原始队列启动一个守护线程进行消息转发
    # 守护线程会在主程序退出时自动终止
    t1 = threading.Thread(target=listen_and_forward, args=(c1,))
    t1.daemon = True
    t1.start()

    t2 = threading.Thread(target=listen_and_forward, args=(c2,))
    t2.daemon = True
    t2.start()

    t_quit = threading.Thread(target=listen_and_forward, args=(quit_channel,))
    t_quit.daemon = True
    t_quit.start()

    # 主循环，从聚合队列中获取消息并处理
    while True:
        which_queue, message = combined_queue.get() # 阻塞等待消息
        if which_queue is c1:
            print('Received value from c1')
        elif which_queue is c2:
            print('Received value from c2')
        elif which_queue is quit_channel:
            print('Received value from quit_channel')
            return # 收到退出信号，终止主循环

if __name__ == '__main__':
    main_direct_simulation()

Received value from c1
Received value from c1
Received value from c2
Received value from c1
Received value from c2
Received value from c1
Received value from c1
Received value from c1
Received value from c1
Received value from c1
Received value from c1
Received value from c1
Received value from quit_channel

import threading
import queue # Python 3 uses queue, Python 2 uses Queue

def select(*queues_to_monitor):
    """
    模拟Go语言的select语句，监听多个Queue对象。

    参数:
        *queues_to_monitor: 可变参数，表示要监听的Queue对象列表。

    返回:
        一个生成器，每次yield (源队列, 消息) 元组。
    """
    combined_queue = queue.Queue(maxsize=0)

    def listen_and_forward(source_queue):
        while True:
            message = source_queue.get()
            combined_queue.put((source_queue, message))

    # 为每个待监听的队列启动一个守护线程
    for q in queues_to_monitor:
        t = threading.Thread(target=listen_and_forward, args=(q,))
        t.daemon = True # 确保线程在主程序退出时自动终止
        t.start()

    # 作为生成器，不断从聚合队列中yield消息
    while True:
        yield combined_queue.get()

def main_with_select_function():
    c1 = queue.Queue(maxsize=0)
    c2 = queue.Queue(maxsize=0)
    quit_channel = queue.Queue(maxsize=0)

    def producer_func1():
        for i in range(10):
            c1.put(i)
        quit_channel.put(0)

    threading.Thread(target=producer_func1).start()

    def producer_func2():
        for i in range(2):
            c2.put(i)

    threading.Thread(target=producer_func2).start()

    # 使用封装好的select函数
    for which_queue, msg in select(c1, c2, quit_channel):
        if which_queue is c1:
            print('Received value from c1')
        elif which_queue is c2:
            print('Received value from c2')
        elif which_queue is quit_channel:
            print('Received value from quit_channel')
            return # 收到退出信号，终止主循环

if __name__ == '__main__':
    main_with_select_function()