在Python中模拟Go语言的select并发模式

package main

import "fmt"

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

    // Goroutine 1: 向 c1 发送数据，最后向 quit 发送信号
    go func() {
        for i := 0; i < 10; i++ {
            c1 <- i
        }
        quit <- 0
    }()

    // Goroutine 2: 向 c2 发送数据
    go func() {
        for i := 0; i < 2; i++ {
            c2 <- i
        }
    }()

    // 主 Goroutine 使用 select 监听多个通道
    for {
        select {
        case <-c1:
            fmt.Println("Received value from c1")
        case <-c2:
            fmt.Println("Received value from c2")
        case <-quit:
            fmt.Println("quit")
            return // 收到 quit 信号后退出
        }
    }
}

import threading
import queue

def main():
    # 模拟Go的通道，使用queue.Queue
    c1 = queue.Queue(maxsize=0)
    c2 = queue.Queue(maxsize=0)
    quit = queue.Queue(maxsize=0)

    # 模拟Go的 Goroutine 1
    def func1():
        for i in range(10):
            c1.put(i)
        quit.put(0) # 完成后发送退出信号

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

    # 模拟Go的 Goroutine 2
    def func2():
        for i in range(2):
            c2.put(i)

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

    # 创建一个中心队列，用于汇总所有通道的消息
    combined = queue.Queue(maxsize=0)

    # 定义一个函数，用于监听指定队列并将消息转发到 combined 队列
    def listen_and_forward(source_queue):
        while True:
            # 获取消息，并将其与源队列一起放入 combined 队列
            combined.put((source_queue, source_queue.get()))

    # 为每个被监听的队列启动一个守护线程进行转发
    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,))
    t_quit.daemon = True
    t_quit.start()

    # 主循环从 combined 队列获取消息
    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:
            print('Received value from quit')
            return # 收到退出信号，主程序终止

if __name__ == '__main__':
    main()

import threading
import queue

def select(*queues):
    """
    模拟Go语言的 select 语句，监听多个 queue.Queue 对象。
    当任一队列有新消息时，yield (源队列, 消息) 元组。
    """
    combined = queue.Queue(maxsize=0)

    def listen_and_forward(source_queue):
        while True:
            combined.put((source_queue, source_queue.get()))

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

    # 作为生成器，持续从 combined 队列中 yield 消息
    while True:
        yield combined.get()

def main_with_select_helper():
    c1 = queue.Queue(maxsize=0)
    c2 = queue.Queue(maxsize=0)
    quit = queue.Queue(maxsize=0)

    def func1():
        for i in range(10):
            c1.put(i)
        quit.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):
        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:
            print('Received value from quit')
            return # 收到退出信号，主程序终止

if __name__ == '__main__':
    main_with_select_helper()