Python中模拟Go语言的Channel 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
        }
    }()

    // 主循环：使用select等待c1, c2或quit通道的消息
    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 # Python 3.x 使用 queue，Python 2.x 使用 Queue

def main_basic_select():
    c1 = queue.Queue(maxsize=0) # 无限大小的队列
    c2 = queue.Queue(maxsize=0)
    quit_q = queue.Queue(maxsize=0)

    # 模拟Go Goroutine 1
    def func1():
        for i in range(10):
            c1.put(i)
        quit_q.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_q = queue.Queue(maxsize=0)

    # 监听并转发消息的函数
    def listen_and_forward(source_queue):
        while True:
            # 从源队列获取消息，并将其与源队列本身一起放入中心队列
            message = source_queue.get()
            combined_q.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_q, message = combined_q.get() # 阻塞直到有消息
        if which_q is c1:
            print(f'Received value from c1: {message}')
        elif which_q is c2:
            print(f'Received value from c2: {message}')
        elif which_q is quit_q:
            print('Received quit signal')
            break # 收到退出信号，终止主循环

if __name__ == '__main__':
    main_basic_select()

import threading
import queue

def select_channels(*queues_to_monitor):
    """
    模拟Go语言的select语句，监听多个Python Queue。
    这是一个生成器函数，每次yield一个(source_queue, message)元组。
    """
    combined_q = queue.Queue(maxsize=0)

    def listen_and_forward(source_queue):
        while True:
            message = source_queue.get()
            combined_q.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_q.get()

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

    # 模拟Go Goroutine 1
    def func1():
        for i in range(10):
            c1.put(i)
        quit_q.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()

    # 使用封装后的select_channels函数
    for which_q, msg in select_channels(c1, c2, quit_q):
        if which_q is c1:
            print(f'Received value from c1: {msg}')
        elif which_q is c2:
            print(f'Received value from c2: {msg}')
        elif which_q is quit_q:
            print('Received quit signal')
            break # 收到退出信号，终止循环

if __name__ == '__main__':
    main_refactored_select()