go - Go: 1 つの生産者と複数の消費者

Question

そのため、Go で 1 つのコンシューマーと多くのプロデューサーを実装する多くの方法を見てきました。これは、 Concurrency in Goの話からの古典的な fanIn 関数です。

私が欲しいのはファンアウト機能です。パラメータとして値を読み取るチャネルを取り、この値のコピーを書き込むチャネルのスライスを返します。

これを実装する正しい/推奨される方法はありますか?

Answer 1

あなたはそれを行うための最良の方法をほとんど説明しましたが、ここにそれを行うコードの小さなサンプルがあります.

遊び場に行く: https://play.golang.org/p/jwdtDXVHJk

package main

import (
    "fmt"
    "time"
)

func producer(iters int) <-chan int {
    c := make(chan int)
    go func() {
        for i := 0; i < iters; i++ {
            c <- i
            time.Sleep(1 * time.Second)
        }
        close(c)
    }()
    return c
}

func consumer(cin <-chan int) {
    for i := range cin {
        fmt.Println(i)
    }
}

func fanOut(ch <-chan int, size, lag int) []chan int {
    cs := make([]chan int, size)
    for i, _ := range cs {
        // The size of the channels buffer controls how far behind the recievers
        // of the fanOut channels can lag the other channels.
        cs[i] = make(chan int, lag)
    }
    go func() {
        for i := range ch {
            for _, c := range cs {
                c <- i
            }
        }
        for _, c := range cs {
            // close all our fanOut channels when the input channel is exhausted.
            close(c)
        }
    }()
    return cs
}

func fanOutUnbuffered(ch <-chan int, size int) []chan int {
    cs := make([]chan int, size)
    for i, _ := range cs {
        // The size of the channels buffer controls how far behind the recievers
        // of the fanOut channels can lag the other channels.
        cs[i] = make(chan int)
    }
    go func() {
        for i := range ch {
            for _, c := range cs {
                c <- i
            }
        }
        for _, c := range cs {
            // close all our fanOut channels when the input channel is exhausted.
            close(c)
        }
    }()
    return cs
}

func main() {
    c := producer(10)
    chans := fanOutUnbuffered(c, 3)
    go consumer(chans[0])
    go consumer(chans[1])
    consumer(chans[2])
}

注意すべき重要な部分は、入力チャネルが使い果たされたときに出力チャネルを閉じる方法です。また、出力チャンネルの 1 つがセンドをブロックすると、他の出力チャンネルのセンドが保留されます。チャネルのバッファ サイズを設定することで、ラグの量を制御します。

Answer 2

以下のこの解決策は少し不自然ですが、私にとってはうまくいきます：

package main

import (
    "fmt"
    "time"
    "crypto/rand"
    "encoding/binary"
)

func handleNewChannels(arrchangen chan [](chan uint32),
                       intchangen chan (chan uint32)) {
    currarr := []chan uint32{}
    arrchangen <- currarr
    for {
        newchan := <-intchangen
        currarr = append(currarr, newchan)
        arrchangen <- currarr
    }
}

func sendToChannels(arrchangen chan [](chan uint32)) {
    tick := time.Tick(1 * time.Second)
    currarr := <-arrchangen
    for {
        select {
        case <-tick:
            sent := false
            var n uint32
            binary.Read(rand.Reader, binary.LittleEndian, &n)
            for i := 0 ; i < len(currarr) ; i++ {
                currarr[i] <- n
                sent = true
            }
            if sent {
                fmt.Println("Sent generated ", n)
            }
        case newarr := <-arrchangen:
            currarr = newarr
        }
    }
}
func handleChannel(tchan chan uint32) {
    for {
        val := <-tchan
        fmt.Println("Got the value ", val)
    }
}

func createChannels(intchangen chan (chan uint32)) {
    othertick := time.Tick(5 * time.Second)
    for {
        <-othertick
        fmt.Println("Creating new channel! ")
        newchan := make(chan uint32)
        intchangen <- newchan
        go handleChannel(newchan)
    }
}

func main() {
    arrchangen := make(chan [](chan uint32))
    intchangen := make(chan (chan uint32))
    go handleNewChannels(arrchangen, intchangen)
    go sendToChannels(arrchangen)
    createChannels(intchangen)
}

Answer 3

各コンシューマのチャネル データのコピーを作成することなく、複数のコンシューマを処理できます。

遊び場に行く: https://play.golang.org/p/yOKindnqiZv

package main

import (
    "fmt"
    "sync"
)

type data struct {
    msg string
    consumers int
}

func main() {
    ch := make(chan *data) // both block or non-block are ok
    var wg sync.WaitGroup
    consumerCount := 3 // specify no. of consumers

    producer := func() {
        obj := &data {
            msg: "hello everyone!",
            consumers: consumerCount,
        }
        ch <- obj
    }
    consumer := func(idx int) {
        defer wg.Done()
        obj := <-ch
        fmt.Printf("consumer %d received data %v\n", idx, obj)
        obj.consumers--
        if obj.consumers > 0 {
            ch <- obj // forward to others
        } else {
            fmt.Printf("last receiver: %d\n", idx)
        }
    }

    go producer()
    for i:=1; i<=consumerCount; i++ {
        wg.Add(1)
        go consumer(i)
    }

    wg.Wait()
}