Wie der broadcast-Nachricht über den Kanal

Eine elegantere Lösung ist ein "broker", wo die Kunden abonnieren können, und unsubscibe Nachrichten.

Zu handhaben auch das abonnieren und abbestellen von Newslettern aus, wir können Kanäle nutzen, so die main-loop des Brokers der erhält und verteilt die Nachrichten integrieren können all diese mit einem einzigen select - Anweisung, und die Synchronisation ist gegeben, von der Lösung der Natur.

Ein weiterer trick ist die Speicherung der Abonnenten in einer map, mapping aus dem Kanal, den wir verwenden zum verteilen von Nachrichten an Sie. Nutzen Sie also den Kanal als Schlüssel in der map, und klicken Sie dann hinzufügen und entfernen des clients ist "tot" einfach. Möglich ist dies, da Kanal-Werte sind vergleichbar, und Ihr Vergleich ist sehr effizient, da-channel-Werte sind einfach Zeiger auf die Kanal-Deskriptoren.

Ohne weitere Umschweife, hier ist eine einfache broker-Implementierung:

type Broker struct {
    stopCh    chan struct{}
    publishCh chan interface{}
    subCh     chan chan interface{}
    unsubCh   chan chan interface{}
}

func NewBroker() *Broker {
    return &Broker{
        stopCh:    make(chan struct{}),
        publishCh: make(chan interface{}, 1),
        subCh:     make(chan chan interface{}, 1),
        unsubCh:   make(chan chan interface{}, 1),
    }
}

func (b *Broker) Start() {
    subs := map[chan interface{}]struct{}{}
    for {
        select {
        case <-b.stopCh:
            return
        case msgCh := <-b.subCh:
            subs[msgCh] = struct{}{}
        case msgCh := <-b.unsubCh:
            delete(subs, msgCh)
        case msg := <-b.publishCh:
            for msgCh := range subs {
                //msgCh is buffered, use non-blocking send to protect the broker:
                select {
                case msgCh <- msg:
                default:
                }
            }
        }
    }
}

func (b *Broker) Stop() {
    close(b.stopCh)
}

func (b *Broker) Subscribe() chan interface{} {
    msgCh := make(chan interface{}, 5)
    b.subCh <- msgCh
    return msgCh
}

func (b *Broker) Unsubscribe(msgCh chan interface{}) {
    b.unsubCh <- msgCh
}

func (b *Broker) Publish(msg interface{}) {
    b.publishCh <- msg
}

Beispiel mit:

func main() {
    //Create and start a broker:
    b := NewBroker()
    go b.Start()

    //Create and subscribe 3 clients:
    clientFunc := func(id int) {
        msgCh := b.Subscribe()
        for {
            fmt.Printf("Client %d got message: %v\n", id, <-msgCh)
        }
    }
    for i := 0; i < 3; i++ {
        go clientFunc(i)
    }

    //Start publishing messages:
    go func() {
        for msgId := 0; ; msgId++ {
            b.Publish(fmt.Sprintf("msg#%d", msgId))
            time.Sleep(300 * time.Millisecond)
        }
    }()

    time.Sleep(time.Second)
}

Ausgabe des obigen wird (versuchen Sie es auf die Gehen Spielplatz):

Client 2 got message: msg#0
Client 0 got message: msg#0
Client 1 got message: msg#0
Client 2 got message: msg#1
Client 0 got message: msg#1
Client 1 got message: msg#1
Client 1 got message: msg#2
Client 2 got message: msg#2
Client 0 got message: msg#2
Client 2 got message: msg#3
Client 0 got message: msg#3
Client 1 got message: msg#3

Verbesserungen

Können Sie prüfen, die folgenden Verbesserungen. Diese können oder möglicherweise nicht nützlich sein, je nachdem, wie /auf was Sie der broker.

Broker.Unsubscribe() können die Meldung schließen, Kanal, der signalisiert, dass keine Nachrichten mehr gesendet werden:

func (b *Broker) Unsubscribe(msgCh chan interface{}) {
    b.unsubCh <- msgCh
    close(msgCh)
}

Dies würde es ermöglichen, Kunden zu range über die Nachricht, Kanal, so wie hier:

msgCh := b.Subscribe()
for msg := range msgCh {
    fmt.Printf("Client %d got message: %v\n", id, msg)
}

Dann, wenn jemand kündigt diese msgCh wie diese:

b.Unsubscribe(msgCh)

Den oben genannten Bereich Schleife wird beendet, nachdem die Verarbeitung aller Nachrichten, die gesendet wurden, bevor der Anruf zu Unsubscribe().

Wenn Sie möchten, dass Ihre Kunden verlassen sich auf die message channel geschlossen wird, und der Makler Lebensdauer ist schmaler als die Ihrer app ist lifetime, dann könnte man auch schließen Sie alle abonnierten clients, wenn der Makler ist angehalten, in der Start() Methode wie diese:

case <-b.stopCh:
    for msgCh := range subs {
        close(msgCh)
    }
    return

InformationsquelleAutor icza

Da Gehen Kanäle Folgen der Communicating Sequential Processes (CSP) - Muster, die Kanäle sind eine Punkt-zu-Punkt-Kommunikation Einheit. Es ist immer ein Schriftsteller und ein Leser beteiligten in jeder exchange.

Jedoch, jeder Kanal Ende werden kann gemeinsamen unter mehreren goroutines. Das ist sicher zu tun - es ist keine gefährliche Rasse Zustand.

Also, es können mehrere Autoren teilen die schreiben Ende. Und/oder es können mehrere Leser teilen die Lesung zu Ende. Ich schrieb diese andere Antwort, die enthält Beispiele.

Wenn Sie wirklich brauchen, eine Sendung, können Sie das direkt tun, aber es ist nicht schwer zu implementieren, eine intermediate-goroutine, kopiert einen Wert aus, um jeweils eine Gruppe der Kanäle.

InformationsquelleAutor Rick-777