comment écouter N canaux? (instruction de sélection dynamique)

pour démarrer une boucle sans fin d'exécution de deux goroutines, je peux utiliser le code ci-dessous:

après avoir reçu le msg, il démarrera une nouvelle goroutine et continuera pour toujours.

c1 := make(chan string)
c2 := make(chan string)

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

for ; true;  {
    select {
    case msg1 := <-c1:
        fmt.Println("received ", msg1)
        go DoStuff(c1, 1)
    case msg2 := <-c2:
        fmt.Println("received ", msg2)
        go DoStuff(c2, 9)
    }
}

Je voudrais maintenant avoir le même comportement pour N goroutines, mais à quoi ressemblera l'instruction select dans ce cas?

C'est le bit de code avec lequel j'ai commencé, mais je ne sais pas comment coder l'instruction select

numChans := 2

//I keep the channels in this slice, and want to "loop" over them in the select statemnt
var chans = [] chan string{}

for i:=0;i<numChans;i++{
    tmp := make(chan string);
    chans = append(chans, tmp);
    go DoStuff(tmp, i + 1)

//How shall the select statment be coded for this case?  
for ; true;  {
    select {
    case msg1 := <-c1:
        fmt.Println("received ", msg1)
        go DoStuff(c1, 1)
    case msg2 := <-c2:
        fmt.Println("received ", msg2)
        go DoStuff(c2, 9)
    }
}

Vous pouvez le faire en utilisant la Selectfonction du reflet package:

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

Select exécute une opération de sélection décrite par la liste des cas. Comme l'instruction Go select, elle se bloque jusqu'à ce qu'au moins un des cas puisse se poursuivre, effectue un choix pseudo-aléatoire uniforme, puis exécute ce cas. Il renvoie l'index du cas choisi et, si ce cas était une opération de réception, la valeur reçue et un booléen indiquant si la valeur correspond à un envoi sur le canal (par opposition à une valeur nulle reçue car le canal est fermé).

Vous transmettez un tableau de SelectCasestructures qui identifient le canal à sélectionner, le sens de l'opération et une valeur à envoyer dans le cas d'une opération d'envoi.

Vous pouvez donc faire quelque chose comme ceci:

cases := make([]reflect.SelectCase, len(chans))
for i, ch := range chans {
    cases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)}
}
chosen, value, ok := reflect.Select(cases)
// ok will be true if the channel has not been closed.
ch := chans[chosen]
msg := value.String()

Vous pouvez expérimenter un exemple plus détaillé ici: http://play.golang.org/p/8zwvSk4kjx

Vous pouvez accomplir cela en enveloppant chaque canal dans un goroutine qui "transfère" les messages vers un canal "agrégé" partagé. Par exemple:

agg := make(chan string)
for _, ch := range chans {
  go func(c chan string) {
    for msg := range c {
      agg <- msg
    }
  }(ch)
}

select {
case msg <- agg:
    fmt.Println("received ", msg)
}

Si vous avez besoin de savoir de quel canal provient le message, vous pouvez l'envelopper dans une structure avec des informations supplémentaires avant de le transmettre au canal agrégé.

Dans mes tests (limités), cette méthode surpasse largement les performances en utilisant le package Reflect:

$ go test dynamic_select_test.go -test.bench=.
...
BenchmarkReflectSelect         1    5265109013 ns/op
BenchmarkGoSelect             20      81911344 ns/op
ok      command-line-arguments  9.463s

Code de référence ici

Pour développer certains commentaires sur les réponses précédentes et fournir une comparaison plus claire, voici un exemple des deux approches présentées jusqu'à présent avec la même entrée, une tranche de canaux à lire et une fonction à appeler pour chaque valeur qui doit également savoir laquelle canal d'où provient la valeur.

Il existe trois différences principales entre les approches:

• Complexité. Bien que cela puisse être en partie une préférence du lecteur, je trouve l'approche du canal plus idiomatique, simple et lisible.

• Performance. Sur mon système Xeon amd64, les canaux goroutines + exécutent la solution de réflexion d'environ deux ordres de grandeur (en général, la réflexion dans Go est souvent plus lente et ne doit être utilisée que lorsque cela est absolument nécessaire). Bien entendu, s'il y a un retard significatif dans la fonction traitant les résultats ou dans l'écriture des valeurs sur les canaux d'entrée, cette différence de performance peut facilement devenir insignifiante.

• Sémantique de blocage / mise en mémoire tampon. L'importance de cela dépend du cas d'utilisation. Le plus souvent, cela n'a pas d'importance ou la légère mise en mémoire tampon supplémentaire dans la solution de fusion de goroutine peut être utile pour le débit. Cependant, s'il est souhaitable d'avoir la sémantique qu'un seul écrivain est débloqué et que sa valeur est entièrement gérée avant qu'un autre écrivain ne soit débloqué, cela ne peut être réalisé qu'avec la solution de réflexion.

Notez que les deux approches peuvent être simplifiées si "l'identifiant" du canal d'envoi n'est pas requis ou si les canaux source ne seront jamais fermés.

Canal de fusion Goroutine:

// Process1 calls `fn` for each value received from any of the `chans`
// channels. The arguments to `fn` are the index of the channel the
// value came from and the string value. Process1 returns once all the
// channels are closed.
func Process1(chans []<-chan string, fn func(int, string)) {
    // Setup
    type item struct {
        int    // index of which channel this came from
        string // the actual string item
    }
    merged := make(chan item)
    var wg sync.WaitGroup
    wg.Add(len(chans))
    for i, c := range chans {
        go func(i int, c <-chan string) {
            // Reads and buffers a single item from `c` before
            // we even know if we can write to `merged`.
            //
            // Go doesn't provide a way to do something like:
            //     merged <- (<-c)
            // atomically, where we delay the read from `c`
            // until we can write to `merged`. The read from
            // `c` will always happen first (blocking as
            // required) and then we block on `merged` (with
            // either the above or the below syntax making
            // no difference).
            for s := range c {
                merged <- item{i, s}
            }
            // If/when this input channel is closed we just stop
            // writing to the merged channel and via the WaitGroup
            // let it be known there is one fewer channel active.
            wg.Done()
        }(i, c)
    }
    // One extra goroutine to watch for all the merging goroutines to
    // be finished and then close the merged channel.
    go func() {
        wg.Wait()
        close(merged)
    }()

    // "select-like" loop
    for i := range merged {
        // Process each value
        fn(i.int, i.string)
    }
}

Réflexion sélectionnez:

// Process2 is identical to Process1 except that it uses the reflect
// package to select and read from the input channels which guarantees
// there is only one value "in-flight" (i.e. when `fn` is called only
// a single send on a single channel will have succeeded, the rest will
// be blocked). It is approximately two orders of magnitude slower than
// Process1 (which is still insignificant if their is a significant
// delay between incoming values or if `fn` runs for a significant
// time).
func Process2(chans []<-chan string, fn func(int, string)) {
    // Setup
    cases := make([]reflect.SelectCase, len(chans))
    // `ids` maps the index within cases to the original `chans` index.
    ids := make([]int, len(chans))
    for i, c := range chans {
        cases[i] = reflect.SelectCase{
            Dir:  reflect.SelectRecv,
            Chan: reflect.ValueOf(c),
        }
        ids[i] = i
    }

    // Select loop
    for len(cases) > 0 {
        // A difference here from the merging goroutines is
        // that `v` is the only value "in-flight" that any of
        // the workers have sent. All other workers are blocked
        // trying to send the single value they have calculated
        // where-as the goroutine version reads/buffers a single
        // extra value from each worker.
        i, v, ok := reflect.Select(cases)
        if !ok {
            // Channel cases[i] has been closed, remove it
            // from our slice of cases and update our ids
            // mapping as well.
            cases = append(cases[:i], cases[i+1:]...)
            ids = append(ids[:i], ids[i+1:]...)
            continue
        }

        // Process each value
        fn(ids[i], v.String())
    }
}

[Code complet sur le terrain de jeu Go .]

Pourquoi cette approche ne fonctionnerait-elle pas en supposant que quelqu'un envoie des événements?

func main() {
    numChans := 2
    var chans = []chan string{}

    for i := 0; i < numChans; i++ {
        tmp := make(chan string)
        chans = append(chans, tmp)
    }

    for true {
        for i, c := range chans {
            select {
            case x = <-c:
                fmt.Printf("received %d \n", i)
                go DoShit(x, i)
            default: continue
            }
        }
    }
}

Option éventuellement plus simple:

Au lieu d'avoir un tableau de canaux, pourquoi ne pas passer un seul canal en tant que paramètre aux fonctions exécutées sur des goroutines séparées, puis écouter le canal dans un goroutine consommateur?

Cela vous permet de sélectionner un seul canal dans votre auditeur, ce qui permet une sélection simple et d'éviter la création de nouvelles goroutines pour regrouper les messages de plusieurs canaux?