PubSub using channels in Go - Eli Bendersky's website

Topic-Based Pub/Sub in Go: Weighing the Design Options

Go's concurrency model encourages structuring programs as goroutines that communicate over channels. The publish-subscribe pattern—also called a message broker or event bus—fits naturally into this model and is frequently used for in-process communication. The goal here is simple: one goroutine produces data and notifies a group of subscriber goroutines, each of which gets its own copy of the data, as opposed to a work queue where messages are consumed on a first-come, first-served basis.

Designing a Pub/Sub system for a Go application involves several non-trivial decisions about channel ownership, buffering, and lifecycle management. We'll walk through a few approaches, starting with the most basic implementation and progressively addressing its shortcomings.

A simple starting point

At its core, the Pub/Sub type needs a registry that maps topic names to a slice of channels, where each channel represents a subscription. The lock protecting this registry is necessary because Go is pragmatic: while the idiom is "share memory by communicating," protecting a shared data structure with a mutex is perfectly reasonable when it keeps the code clear.

type Pubsub struct {
  mu   sync.RWMutex
  subs map[string][]chan string
}

The constructor is straightforward, and the Subscribe method takes advantage of Go's default value semantics. If a topic has no existing subscribers, indexing ps.subs returns an empty slice, which can be appended to directly:

func NewPubsub() *Pubsub {
  ps := &Pubsub{}
  ps.subs = make(map[string][]chan string)
  return ps
}

func (ps *Pubsub) Subscribe(topic string, ch chan string) {
  ps.mu.Lock()
  defer ps.mu.Unlock()

  ps.subs[topic] = append(ps.subs[topic], ch)
}

Similarly, Publish iterates over the subscribers for a topic. If none exist, the loop simply doesn't execute:

func (ps *Pubsub) Publish(topic string, msg string) {
  ps.mu.RLock()
  defer ps.mu.RUnlock()

  for _, ch := range ps.subs[topic] {
    ch <- msg
  }
}

This version is functional but incomplete. It lacks an Unsubscribe method, and more importantly, it never closes subscription channels. In Go, closing a channel is how the sender signals that no more messages will arrive. Without it, subscribers have no way to know when to stop listening and clean up resources.

Closing subscription channels

To address this, we can add a closed flag to the Pubsub struct, initialize it to false in the constructor, and modify Publish to check it before sending. A new Close method then signals to all subscribers that the stream is ending:

type Pubsub struct {
  mu     sync.RWMutex
  subs   map[string][]chan string
  closed bool
}

func (ps *Pubsub) Publish(topic string, msg string) {
  ps.mu.RLock()
  defer ps.mu.RUnlock()

  if ps.closed {
    return
  }

  for _, ch := range ps.subs[topic] {
    ch <- msg
  }
}

func (ps *Pubsub) Close() {
  ps.mu.Lock()
  defer ps.mu.Unlock()

  if !ps.closed {
    ps.closed = true
    for _, subs := range ps.subs {
      for _, ch := range subs {
        close(ch)
      }
    }
  }
}

This design raises an important question about channel ownership. The channels are created by the client and passed to Subscribe, but Close is now responsible for closing them. In idiomatic Go, the sending side should close a channel to signal completion—closing on the receiving side is dangerous because the sender may not know and could panic if it tries to send on a closed channel.

Buffering and blocking

The more critical issue is blocking. Consider the send loop in Publish:

for _, ch := range ps.subs[topic] {
  ch <- msg
}

If a subscription channel is unbuffered, the send will block until the receiver consumes the message. This stalls not only that subscriber but also all other subscribers on the same topic. If receivers are slow, unbuffered channels create a bottleneck.

Buffering helps, but in this design the buffer size is decided by the client who creates the channel. This shifts responsibility to the client and means Pub/Sub's performance depends on its users' choices. A single client passing an unbuffered channel can indirectly block everyone else on the topic.

Pub/Sub creates its own channels

An alternative is to have Pubsub create subscription channels itself. Only the Subscribe method changes: instead of accepting a channel, it creates a buffered channel (with a size of 1 in this example) and returns it:

func (ps *Pubsub) Subscribe(topic string) <-chan string {
  ps.mu.Lock()
  defer ps.mu.Unlock()

  ch := make(chan string, 1)
  ps.subs[topic] = append(ps.subs[topic], ch)
  return ch
}

This arrangement centralizes channel ownership. Pubsub creates, writes to, and closes the channels, making the subscriber's life cycle clearer: Subscribe returns a channel that can be read until it is closed. Configuring the buffer size per subscription or for all subscriptions in the constructor is a simple extension.

The main inconvenience is that a client can no longer subscribe the same channel to multiple topics. In the previous version, passing the same channel to several Subscribe calls was possible, but that practice creates its own hazards. Close could panic by closing the same channel twice; avoiding that requires tracking a set of already-closed channels. The cleaner path is one channel per subscription, using fan-in patterns to merge topics when a single consumer loop is desired.

Spawning a goroutine per send

To sidestep blocking entirely, one could run each send in its own goroutine:

func (ps *Pubsub) Publish(topic string, msg string) {
  ps.mu.RLock()
  defer ps.mu.RUnlock()

  if ps.closed {
    return
  }

  for _, ch := range ps.subs[topic] {
    go func(ch chan string) {
      ch <- msg
    }(ch)
  }
}

This eliminates the buffering dependency—no channel send blocks another send. But it introduces two problems. First, spawning a goroutine per message has a performance cost. It's small, but at high message rates it can add up; benchmarking is the only way to know if it matters for a particular workload.

Second, and more seriously, this design decouples sends from channel closure. A slow client whose channel is blocked for a long time could race with Close, which may attempt to close the channel while a send is still pending. Closing a channel with pending writes is a race condition—among the hardest bugs to diagnose. In the synchronous version, this cannot happen because Publish holds the lock until all sends complete, preventing Close from running concurrently.

Weighing the options

Each version of this small Pub/Sub example highlights a different trade-off. The version where Subscribe creates and returns its own buffered channels is the most conceptually sound option. It centralizes ownership: Pubsub creates the channels, writes to them, and closes them. The client sees a simple contract—the channel comes from Subscribe and is readable until it closes. Adding configurable buffering is trivial, and the design avoids the race conditions that appear when sends are split across goroutines.

Channels are powerful in Go, but they are not magic. Ownership, blocking, and cleanup are still design problems that require careful thought, and this exercise shows how the same small API can be shaped in several different directions depending on the priorities of the application.