Python中模拟Go语言的select语句：实现多通道并发通信

package main

import "fmt"

func main() {
    c1 := make(chan int)
    c2 := make(chan int)
    quit := make(chan int)

    go func() {
        for i := 0; i < 10; i++ {
            c1 <- i
        }
        quit <- 0 // 完成后发送退出信号
    }()

    go func() {
        for i := 0; i < 2; i++ {
            c2 <- i
        }
    }()

    for {
        select {
        case <-c1:
            fmt.Println("Received value from c1")
        case <-c2:
            fmt.Println("Received value from c2")
        case <-quit:
            fmt.Println("quit")
            return // 收到退出信号后终止循环
        }
    }
}

import threading
import queue

def main_direct_translation():
    c1 = queue.Queue(maxsize=0) # 无界队列
    c2 = queue.Queue(maxsize=0)
    quit_q = queue.Queue(maxsize=0) # 使用quit_q避免与Python内置关键字冲突

    # 生产者线程1：向 c1 发送数据，完成后向 quit_q 发送退出信号
    def func1():
        for i in range(10):
            c1.put(i)
        quit_q.put(0)

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

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

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

    # 聚合队列，用于接收来自所有通道的消息
    combined = queue.Queue(maxsize=0)

    # 转发器函数：监听一个队列，并将消息转发到聚合队列
    def listen_and_forward(source_queue):
        while True:
            # 获取消息，并将其与来源队列一起放入聚合队列
            message = source_queue.get()
            combined.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_q,))
    t_quit.daemon = True
    t_quit.start()

    # 主循环：从聚合队列中获取消息并处理
    while True:
        which_queue, message = combined.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_q:
            print('Received value from quit')
            return # 收到退出信号，终止主循环

main_direct_translation()

import threading
import queue

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

    参数:
        *queues: 任意数量的queue.Queue对象。

    生成器:
        每次有消息到达时，yield一个元组 (source_queue, message)，
        其中 source_queue 是消息的来源队列，message 是接收到的消息。
    """
    combined = queue.Queue(maxsize=0)

    # 转发器函数：监听单个队列，并将消息转发到聚合队列
    def listen_and_forward(source_queue):
        while True:
            try:
                message = source_queue.get()
                combined.put((source_queue, message))
            except Exception as e:
                # 实际应用中可能需要更健壮的错误处理，例如处理队列关闭
                print(f"Error in forwarder thread for {source_queue}: {e}")
                break # 退出循环，终止线程

    # 为每个传入的队列启动一个转发器线程
    for q in queues:
        t = threading.Thread(target=listen_and_forward, args=(q,))
        t.daemon = True
        t.start()

    # 主循环：从聚合队列中获取消息并作为生成器结果返回
    while True:
        yield combined.get()

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

    def func1():
        for i in range(10):
            c1.put(i)
        quit_q.put(0)

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

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

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

    # 使用封装后的 select 函数
    for which_queue, msg in select(c1, c2, quit_q):
        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_q:
            print('Received value from quit')
            return # 收到退出信号，终止主循环

main_with_select_function()