go并发模式 or-do-channel + bridge

package main

import (
    "context"
    "fmt"
)
// orDone
func orDone(ctx context.Context, value <-chan int) <-chan int {
    ordoneStream := make(chan int)
    go func() {
        defer close(ordoneStream)
        for {
            select {
            case <-ctx.Done():
                return
            case v, ok := <-value:
                if !ok {
                    return
                }
                select {
                case <-ctx.Done():
                    return
                case ordoneStream <- v:
                }
            }
        }
    }()
    return ordoneStream
}

// bridge
func bridge(ctx context.Context, value <-chan <-chan int) <-chan int {
    bridgeStream := make(chan int)
    go func() {
        defer close(bridgeStream)
        for {
            var stream <-chan int
            select {
            case <-ctx.Done():
                return
            case v, ok := <-value:
                if !ok {
                    return
                }
                stream = v
            }
            for i := range orDone(ctx, stream) {
                select {
                case <-ctx.Done():
                    return
                case bridgeStream <- i:
                }
            }
        }
    }()
    return bridgeStream
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    ch1 := make(chan int)
    ch2 := make(chan int)
    ch3 := make(chan int)
    stream := make(chan (<-chan int))

    go func() {
        defer close(stream)
        stream <- ch1
        stream <- ch2
        stream <- ch3
    }()

    go func() {
        defer close(ch1)
        for i := 0; i < 3; i++ {
            ch1 <- i + 1
        }
    }()

    go func() {
        defer close(ch2)
        for i := 0; i < 3; i++ {
            ch2 <- i + 11
        }
    }()

    go func() {
        defer close(ch3)
        for i := 0; i < 3; i++ {
            ch3 <- i + 21
        }
    }()

    for val := range bridge(ctx, stream) {
        fmt.Println(val)
    }

}

 【版权申明】未经博主同意,谢绝转载!(请尊重原创,博主保留追究权) https://www.cnblogs.com/facetwitter/p/18293865