Go: Один производитель многих потребителей

Ответ 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])
}

Важная часть, которую следует отметить, заключается в том, как мы закрываем выходные каналы после исчерпания входного канала. Также, если один из выходных каналов блокируется при отправке, он будет удерживать передачу на других выходных каналах. Мы контролируем величину задержки, задавая размер буфера каналов.

Ответ 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)
}

Ответ 3

Во-первых, см. связанный с ним вопрос Какая самая необычная идиома для производителя/потребителя в Go? и Один поток, проявляющий интерес к другой поток (потребитель/производитель). Также обратите внимание на проблема производителя-потребителя. О concurrency см. как достичь concurrency в Google Go.