Go语言中实现动态N个通道的select操作

package main

import (
    "fmt"
    "time"
    "strconv"
)

// 模拟一个向通道发送数据的goroutine
func DoStuff(c chan string, id int, delay int) {
    for i := 0; i < 3; i++ { // 模拟发送几次数据
        time.Sleep(time.Duration(delay) * time.Second)
        msg := fmt.Sprintf("Channel %d: Message %d", id, i+1)
        c <- msg
    }
    close(c) // 发送完毕后关闭通道
}

func main() {
    c1 := make(chan string)
    c2 := make(chan string)

    go DoStuff(c1, 1, 5)
    go DoStuff(c2, 2, 2)

    // 静态select语句，只能处理固定数量的通道
    for {
        select {
        case msg1, ok := <-c1:
            if !ok {
                c1 = nil // 通道关闭后设置为nil，避免再次选择
                fmt.Println("Channel 1 closed.")
                break
            }
            fmt.Println("Received from c1:", msg1)
            // 根据需求，可能在此处启动新的goroutine或重新安排任务
            // go DoStuff(c1, 1, 1) // 示例中表示可能重新启动
        case msg2, ok := <-c2:
            if !ok {
                c2 = nil
                fmt.Println("Channel 2 closed.")
                break
            }
            fmt.Println("Received from c2:", msg2)
            // go DoStuff(c2, 2, 9) // 示例中表示可能重新启动
        }

        // 如果所有通道都已关闭，则退出循环
        if c1 == nil && c2 == nil {
            fmt.Println("All channels closed, exiting.")
            break
        }
    }
}

func Select(cases []SelectCase) (chosen int, recv Value, recvOK bool)

type SelectCase struct {
    Dir  SelectDir   // 操作方向：SelectRecv（接收）、SelectSend（发送）或 SelectDefault（默认）
    Chan Value       // 通道，需要通过reflect.ValueOf包装
    Send Value       // 如果是发送操作，则为要发送的值
}

import "reflect"

// ... (假设chans已经初始化并包含了N个通道)
var chans = []chan string{ /* ... 动态创建的通道 ... */ }

cases := make([]reflect.SelectCase, len(chans))
for i, ch := range chans {
    cases[i] = reflect.SelectCase{
        Dir:  reflect.SelectRecv,        // 指定为接收操作
        Chan: reflect.ValueOf(ch),       // 将通道包装为reflect.Value
    }
}

package main

import (
    "fmt"
    "reflect"
    "strconv"
    "sync"
    "time"
)

// DoStuff 模拟一个向通道发送数据的goroutine
// 参数c: 要发送数据的通道
// 参数id: goroutine的唯一标识
// 参数delay: 每次发送数据前的延迟时间（秒）
func DoStuff(c chan string, id int, delay int, wg *sync.WaitGroup) {
    defer wg.Done() // goroutine结束时通知WaitGroup

    for i := 0; i < 3; i++ { // 模拟发送几次数据
        time.Sleep(time.Duration(delay) * time.Second)
        msg := fmt.Sprintf("Channel %d: Message %d (Delay: %ds)", id, i+1, delay)
        select {
        case c <- msg:
            // 成功发送
        case <-time.After(100 * time.Millisecond): // 防止阻塞过久
            fmt.Printf("DoStuff %d: Send timed out for msg %d\n", id, i+1)
            return // 如果发送超时，则退出
        }
    }
    // 注意：这里不再关闭通道，因为通道可能被主循环重新用于新的DoStuff任务
    // 如果每个DoStuff任务都关闭通道，那么主循环在处理已关闭通道时需要额外的逻辑
    // 在这种动态重用的场景下，通常由外部统一管理通道的生命周期或在特定条件下关闭。
}

func main() {
    numChans := 3 // 动态设置通道数量
    var chans = make([]chan string, numChans)
    var activeChans = make(map[int]struct{}) // 记录活跃通道的索引

    // 初始化通道和启动初始的DoStuff goroutine
    var initialWg sync.WaitGroup
    for i := 0; i < numChans; i++ {
        chans[i] = make(chan string)
        initialWg.Add(1)
        go DoStuff(chans[i], i+1, (i+1)*2, &initialWg) // 不同的延迟
        activeChans[i] = struct{}{} // 标记为活跃
    }
    // 等待所有初始的DoStuff goroutine发送完第一批消息
    // 实际应用中可能不需要等待，而是直接进入主循环
    // initialWg.Wait() // 如果不等待，主循环可能会在DoStuff还没发送数据时就执行reflect.Select

    fmt.Println("Starting dynamic channel listener...")

    // 主循环：动态监听所有通道
    var listenerWg sync.WaitGroup // 用于等待所有重新启动的DoStuff goroutine
    for {
        if len(activeChans) == 0 {
            fmt.Println("All active channels processed, exiting main loop.")
            break
        }

        // 1. 构建reflect.SelectCase切片
        // 每次循环都需要重新构建，以处理通道状态变化（例如被设置为nil）
        var currentCases []reflect.SelectCase
        var currentChanIndices []int // 记录cases中对应的原始通道索引

        for i := 0; i < numChans; i++ {
            if _, isActive := activeChans[i]; isActive { // 只处理活跃的通道
                // 确保通道不为nil，如果通道已关闭并处理完毕，可以将其从activeChans中移除
                if chans[i] != nil {
                    currentCases = append(currentCases, reflect.SelectCase{
                        Dir:  reflect.SelectRecv,
                        Chan: reflect.ValueOf(chans[i]),
                    })
                    currentChanIndices = append(currentChanIndices, i) // 记录原始索引
                }
            }
        }

        if len(currentCases) == 0 {
            fmt.Println("No active channels to listen to. Waiting for any pending goroutines...")
            listenerWg.Wait() // 等待所有重新启动的DoStuff完成
            break
        }

        // 2. 执行reflect.Select
        chosen, value, ok := reflect.Select(currentCases)

        // 3. 处理选择结果
        originalChanIndex := currentChanIndices[chosen] // 获取被选中通道的原始索引
        chosenChan := chans[originalChanIndex]          // 获取被选中通道

        if ok { // 通道未关闭，成功接收到消息
            msg := value.String()
            fmt.Printf("Received from Channel %d (original index %d): %s\n", originalChanIndex+1, originalChanIndex, msg)

            // 模拟原始问题中的行为：接收消息后启动一个新的DoStuff goroutine
            // 这会使得该通道再次活跃起来，继续发送数据
            listenerWg.Add(1)
            go DoStuff(chosenChan, originalChanIndex+1, (originalChanIndex+1)*3, &listenerWg) // 重新启动，使用新的延迟
        } else { // 通道已关闭
            fmt.Printf("Channel %d (original index %d) closed.\n", originalChanIndex+1, originalChanIndex)
            // 将已关闭的通道从活跃列表中移除
            delete(activeChans, originalChanIndex)
            // 也可以选择将chans[originalChanIndex]设置为nil，避免在后续循环中重复处理
            // chans[originalChanIndex] = nil // 如果DoStuff不再向此通道发送，可以这样做
        }
        time.Sleep(100 * time.Millisecond) // 短暂暂停，避免CPU空转过高
    }

    // 等待所有因接收消息而重新启动的DoStuff goroutine完成
    fmt.Println("Waiting for all re-spawned DoStuff goroutines to finish...")
    listenerWg.Wait()
    fmt.Println("Program finished.")
}