go并发模式 扇入扇出

扇入扇出寻找素数：

package main

import (
    "fmt"
    "math/rand"
    "runtime"
    "sync"
    "time"
)

var repeatFn = func(done <-chan interface{}, fn func() interface{}) <-chan interface{} {
    valueStream := make(chan interface{})
    go func() {
        defer close(valueStream)
        for {
            select {
            case <-done:
                return
            case valueStream <- fn():
            }
        }
    }()
    return valueStream
}

var toInt = func(done <-chan interface{}, valueStream <-chan interface{}) <-chan int {
    intStream := make(chan int)
    go func() {
        defer close(intStream)
        for v := range valueStream {
            select {
            case <-done:
                return
            case intStream <- v.(int):
            }
        }
    }()
    return intStream
}

var primeChanArr = func(done <-chan interface{}, intStream <-chan int) <-chan int {
    primeStream := make(chan int)
    go func() {
        defer close(primeStream)
        for num := range intStream {
            num -= 1
            pirm := true
            for divisor := num - 1; divisor > 1; divisor-- {
                if num%divisor == 0 {
                    pirm = false
                    break
                }
            }
            if pirm {
                select {
                case <-done:
                    return
                case primeStream <- num:
                }
            }
        }
    }()
    return primeStream
}

var take = func(done <-chan interface{}, valueStream <-chan int, num int) <-chan int {
    takeStream := make(chan int)
    go func() {
        defer close(takeStream)
        for i := 0; i < num; i++ {
            select {
            case <-done:
                return
            case v := <-valueStream:
                takeStream <- v
            }
        }
    }()
    return takeStream
}

var fanIn = func(done <-chan interface{}, channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    faninStream := make(chan int)
    multiplex := func(c <-chan int) {
        defer wg.Done()
        for i := range c {
            select {
            case <-done:
                return
            case faninStream <- i:
            }
        }
    }

    wg.Add(len(channels))
    for _, x := range channels {
        go multiplex(x)
    }

    go func() {
        wg.Wait()
        close(faninStream)
    }()

    return faninStream
}

func main() {
    randNum := func() interface{} { return rand.Intn(50000000) }
    done := make(chan interface{})
    defer close(done)
    start := time.Now()

    randIntSream := toInt(done, repeatFn(done, randNum))
    cpuNumber := runtime.NumCPU()
    fmt.Printf("CPU Number: %d\n", cpuNumber)

    chanArray := make([]<-chan int, cpuNumber)

    fmt.Printf("Primes：\n")

    for i := 0; i < cpuNumber; i++ {
        chanArray[i] = primeChanArr(done, randIntSream)
    }

    for prime := range take(done, fanIn(done, chanArray...), 10) {
        fmt.Printf("%d\n", prime)
    }
    fmt.Printf("耗时：%v\n", time.Since(start))
}

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