Pub Sub in Go With Channels and Goroutines From Scratch

Publish and subscribe is one of those patterns that every Go program ends up reinventing. A config change should reach the cache and the logger. A user signing in should reach the audit trail and the metrics. Something produces events, several somethings consume them, and the producer should not know who they are. In one process, channels give you almost everything you need, and the remaining parts are small but easy to get wrong: a slow subscriber, a subscriber that leaves while a message is in flight, and shutdown.

This project builds a small generic broker step by step, with the decisions that matter called out, and tests that pin them down. It belongs to the concurrency patterns cluster and uses channels, the mutex and context. Before building one, read why most event buses are too clever, because a direct call is often better. Everything ran on Go 1.25.6.

What we are building

The requirements, written down before any code:

1. Publishers send a message to a named topic. They never wait for subscribers.
2. Any number of subscribers can listen to a topic, and each receives every message published after it subscribed.
3. One slow subscriber must not delay the publisher or the other subscribers.
4. A subscriber can leave at any time, including while a message is being published, without a panic.
5. Closing the broker ends every subscription cleanly.
6. The type of message is checked by the compiler.

Requirement 3 is the one that shapes the design. If all subscribers shared one channel, a slow one would hold everyone back, and the messages would be split among them instead of copied to each. So every subscriber gets its own channel, and the broker copies each message into all of them.

The API

package main

import (
	"sync"
	"sync/atomic"
)

// Subscription is one subscriber's view of a topic.
type Subscription[T any] struct {
	// C delivers messages. It is closed when the subscription ends.
	C <-chan T

	ch      chan T
	topic   string
	broker  *Broker[T]
	dropped atomic.Int64
}

// Dropped reports how many messages were discarded because C was full.
func (s *Subscription[T]) Dropped() int64 { return s.dropped.Load() }

// Cancel ends the subscription and closes C. It is safe to call more than once.
func (s *Subscription[T]) Cancel() { s.broker.remove(s) }

// Broker fans messages out to the subscribers of each topic.
type Broker[T any] struct {
	mu     sync.RWMutex
	topics map[string]map[*Subscription[T]]struct{}
	closed bool
}

// NewBroker returns an empty broker.
func NewBroker[T any]() *Broker[T] {
	return &Broker[T]{topics: make(map[string]map[*Subscription[T]]struct{})}
}

A Subscription carries the read-only channel C that callers use, and the writable ch that the broker uses. They are the same channel; the split means a subscriber cannot send on it or close it by accident, because the type of C forbids both. The dropped counter is an atomic.Int64, because the publisher increments it from its own goroutine while the subscriber might read it from another.

Subscribe

package main

// Subscribe registers for topic. buffer is how many messages may wait unread; a subscriber
// that falls further behind than that starts losing messages (see Publish).
// On a closed broker it returns a subscription whose channel is already closed.
func (b *Broker[T]) Subscribe(topic string, buffer int) *Subscription[T] {
	ch := make(chan T, buffer)
	s := &Subscription[T]{C: ch, ch: ch, topic: topic, broker: b}

	b.mu.Lock()
	defer b.mu.Unlock()
	if b.closed {
		close(ch)
		return s
	}
	if b.topics[topic] == nil {
		b.topics[topic] = make(map[*Subscription[T]]struct{})
	}
	b.topics[topic][s] = struct{}{}
	return s
}

The buffer size is the caller's decision because only the caller knows how bursty its own consumption is. A subscriber that does a database write per message wants a larger buffer than one that increments a counter. Subscribing to a closed broker returns an already-closed channel instead of an error, so a consumer's for msg := range sub.C loop ends immediately, which is the right behavior for a service that is shutting down.

Publish, and the slow subscriber policy

The publish method has three jobs: find the subscribers, deliver without blocking, and account for what it could not deliver.

package main

// Publish copies msg to every subscriber of topic without blocking. It returns how many
// subscribers received it and how many had a full buffer and missed it.
func (b *Broker[T]) Publish(topic string, msg T) (delivered, dropped int) {
	b.mu.RLock()
	defer b.mu.RUnlock()
	for s := range b.topics[topic] {
		select {
		case s.ch <- msg:
			delivered++
		default:
			s.dropped.Add(1)
			dropped++
		}
	}
	return delivered, dropped
}

The select with a default is the non-blocking send. If the subscriber's buffer has room the message goes in; if not, the message is dropped for that subscriber and counted, and the loop moves on to the next one. The policy here is drop the newest: a full buffer keeps the older messages and loses the arriving one. That is a choice. Three are common:

• Drop the newest (this code). Cheap, bounded memory, and the subscriber sees an old but contiguous prefix. Good for events where later ones supersede earlier ones only if you replace rather than queue, which this does not, so it suits "best effort" feeds such as metrics.
• Drop the oldest. The subscriber always has the latest state, at the cost of a gap in the middle. It needs a receive-and-resend inside the broker and a little care to stay race-free. Good for "current value" feeds such as a dashboard.
• Block, with a deadline. The publisher waits for the slow subscriber up to a limit. It gives lossless delivery to subscribers who keep up and makes the publisher as slow as the slowest. Good when every message matters and subscribers are trusted to be quick, and the wrong choice for anything exposed to untrusted consumers.

The counter lets an operator see the policy at work: a subscriber with a rising Dropped() needs a bigger buffer, faster handling, or a different policy. Silent loss is the worst outcome, so count it. The broker holds the read lock across the sends. That is acceptable because every send is non-blocking and so the lock is held for microseconds, and it is what makes the next part safe.

Unsubscribe without a panic

Sending on a closed channel panics. The nightmare scenario is a subscriber cancelling at the instant the broker is sending to it. Two rules prevent it. The broker only sends while holding the read lock, and it only closes a subscriber's channel while holding the write lock. The two never overlap, so a send cannot happen on a channel that is mid-close:

package main

func (b *Broker[T]) remove(s *Subscription[T]) {
	b.mu.Lock()
	defer b.mu.Unlock()
	subs, ok := b.topics[s.topic]
	if !ok {
		return
	}
	if _, found := subs[s]; !found {
		return // already cancelled
	}
	delete(subs, s)
	if len(subs) == 0 {
		delete(b.topics, s.topic)
	}
	close(s.ch)
}

// Close ends every subscription and refuses new ones. Publishing afterwards delivers nothing.
func (b *Broker[T]) Close() {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.closed {
		return
	}
	b.closed = true
	for _, subs := range b.topics {
		for s := range subs {
			close(s.ch)
		}
	}
	b.topics = make(map[string]map[*Subscription[T]]struct{})
}

remove checks that the subscription is still registered before closing, which makes Cancel idempotent: a second call finds nothing and returns. That covers the common case of a deferred Cancel after Close has already ended everything. This is the same discipline as closing a channel: one party owns the close, and here it is always the broker.

Blocking publish with a context

Some callers want lossless delivery and are willing to wait. A second method takes a context and blocks on each subscriber until the message fits or the context ends. It cannot hold the read lock while waiting, since a blocked subscriber would then block Cancel and Close behind it. So it takes a snapshot of the subscriber list under the lock and sends outside it, accepting that a subscriber may be cancelled in between:

package main

import "context"

// PublishWait delivers msg to every subscriber, waiting for room in each buffer until ctx ends.
// A subscriber that cancels while we wait is skipped.
func (b *Broker[T]) PublishWait(ctx context.Context, topic string, msg T) error {
	b.mu.RLock()
	snapshot := make([]*Subscription[T], 0, len(b.topics[topic]))
	for s := range b.topics[topic] {
		snapshot = append(snapshot, s)
	}
	b.mu.RUnlock()

	for _, s := range snapshot {
		if err := b.sendOne(ctx, s, msg); err != nil {
			return err
		}
	}
	return nil
}

func (b *Broker[T]) sendOne(ctx context.Context, s *Subscription[T], msg T) (err error) {
	// The channel may be closed between the snapshot and this send. That is a normal race
	// with Cancel, so treat the resulting panic as "this subscriber is gone".
	defer func() {
		if r := recover(); r != nil {
			err = nil
		}
	}()
	select {
	case s.ch <- msg:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

I do not love the recover in sendOne. It exists because no lock can protect a blocking send without also blocking cancellation, and the choice is between a recover for the one specific case of a send racing a close, or giving every subscription a done channel that the send also selects on. The second is cleaner and costs a channel per subscription; for this article the recover keeps the code short, and I would switch to a per-subscription done channel if the broker were going into a library. Note that recovering from a closed-channel send is one of the few legitimate uses of recover for a runtime error, because the cause is known and the state is fine afterwards (recover).

A demo you can run

Three subscribers, two topics. One of the "orders" subscribers is slow, with a buffer of 2, and does not read until the publisher is finished. The other has a buffer of 10 and drains everything:

package main

import (
	"fmt"
	"sync"
)

func main() {
	b := NewBroker[string]()

	fast := b.Subscribe("orders", 10)
	slow := b.Subscribe("orders", 2)
	alerts := b.Subscribe("alerts", 4)

	for i := 1; i <= 5; i++ {
		delivered, dropped := b.Publish("orders", fmt.Sprintf("order-%d", i))
		fmt.Printf("order-%d: delivered to %d, dropped for %d\n", i, delivered, dropped)
	}
	b.Publish("alerts", "disk 91% full")

	var wg sync.WaitGroup
	read := func(name string, s *Subscription[string]) {
		defer wg.Done()
		var got []string
		for {
			select {
			case m, ok := <-s.C:
				if !ok {
					fmt.Printf("%-6s got %v (dropped %d)\n", name, got, s.Dropped())
					return
				}
				got = append(got, m)
			default:
				fmt.Printf("%-6s got %v (dropped %d)\n", name, got, s.Dropped())
				return
			}
		}
	}
	wg.Add(3)
	read("fast", fast)
	read("slow", slow)
	read("alerts", alerts)
	wg.Wait()

	b.Close()
	_, open := <-fast.C
	fmt.Println("channel open after Close:", open)
}

order-1: delivered to 2, dropped for 0
order-2: delivered to 2, dropped for 0
order-3: delivered to 1, dropped for 1
order-4: delivered to 1, dropped for 1
order-5: delivered to 1, dropped for 1
fast   got [order-1 order-2 order-3 order-4 order-5] (dropped 0)
slow   got [order-1 order-2] (dropped 3)
alerts got [disk 91% full] (dropped 0)
channel open after Close: false

The publisher was never blocked, even though the slow subscriber stopped accepting after two messages: the first two orders went to both subscribers, and from the third on, the slow one dropped, so the fast one received all five and the slow one received order-1 and order-2 with three drops counted. The alerts subscriber saw only its own topic. After Close the channel is closed and drained, which the final line confirms.

Tests that pin the behavior

A broker is a good candidate for tests, because the properties are crisp and the bugs are the kind that only show up under concurrency. These cover the six requirements, run under -race:

package main

import (
	"context"
	"sync"
	"testing"
	"time"
)

func TestFanOutDeliversToEverySubscriber(t *testing.T) {
	b := NewBroker[int]()
	defer b.Close()
	a, c := b.Subscribe("n", 4), b.Subscribe("n", 4)
	b.Publish("n", 1)
	b.Publish("n", 2)
	for _, s := range []*Subscription[int]{a, c} {
		if x, y := <-s.C, <-s.C; x != 1 || y != 2 {
			t.Fatalf("want 1,2 in order, got %d,%d", x, y)
		}
	}
}

func TestSlowSubscriberDoesNotBlockOthers(t *testing.T) {
	b := NewBroker[int]()
	defer b.Close()
	slow := b.Subscribe("n", 1)
	fast := b.Subscribe("n", 100)
	done := make(chan struct{})
	go func() {
		for i := 0; i < 50; i++ {
			b.Publish("n", i)
		}
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("publisher blocked on a slow subscriber")
	}
	if got := len(fast.C); got != 50 {
		t.Fatalf("fast subscriber has %d messages, want 50", got)
	}
	if slow.Dropped() != 49 {
		t.Fatalf("slow dropped %d, want 49", slow.Dropped())
	}
}

func TestCancelIsIdempotentAndClosesChannel(t *testing.T) {
	b := NewBroker[int]()
	s := b.Subscribe("n", 1)
	s.Cancel()
	s.Cancel()
	if _, open := <-s.C; open {
		t.Fatal("channel should be closed after Cancel")
	}
	if d, _ := b.Publish("n", 1); d != 0 {
		t.Fatalf("delivered %d after cancel, want 0", d)
	}
}

func TestCancelWhilePublishing(t *testing.T) {
	b := NewBroker[int]()
	defer b.Close()
	var wg sync.WaitGroup
	for range 8 {
		wg.Go(func() {
			for range 200 {
				s := b.Subscribe("n", 1)
				s.Cancel()
			}
		})
	}
	wg.Go(func() {
		for i := 0; i < 20000; i++ {
			b.Publish("n", i)
		}
	})
	wg.Wait()
}

func TestCloseEndsEverySubscription(t *testing.T) {
	b := NewBroker[int]()
	s := b.Subscribe("n", 1)
	b.Close()
	if _, open := <-s.C; open {
		t.Fatal("channel should be closed after broker Close")
	}
	late := b.Subscribe("n", 1)
	if _, open := <-late.C; open {
		t.Fatal("subscribing to a closed broker should return a closed channel")
	}
}

func TestPublishWaitHonoursContext(t *testing.T) {
	b := NewBroker[int]()
	defer b.Close()
	b.Subscribe("n", 1) // nobody reads it
	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	defer cancel()
	if err := b.PublishWait(ctx, "n", 1); err != nil {
		t.Fatalf("first message fits in the buffer, got %v", err)
	}
	if err := b.PublishWait(ctx, "n", 2); err != context.DeadlineExceeded {
		t.Fatalf("second message should time out, got %v", err)
	}
}

=== RUN   TestFanOutDeliversToEverySubscriber
--- PASS: TestFanOutDeliversToEverySubscriber (0.00s)
=== RUN   TestSlowSubscriberDoesNotBlockOthers
--- PASS: TestSlowSubscriberDoesNotBlockOthers (0.00s)
=== RUN   TestCancelIsIdempotentAndClosesChannel
--- PASS: TestCancelIsIdempotentAndClosesChannel (0.00s)
=== RUN   TestCancelWhilePublishing
--- PASS: TestCancelWhilePublishing (0.03s)
=== RUN   TestCloseEndsEverySubscription
--- PASS: TestCloseEndsEverySubscription (0.00s)
=== RUN   TestPublishWaitHonoursContext
--- PASS: TestPublishWaitHonoursContext (0.05s)
PASS
ok  	example.test/lab	1.341s

The test I would not skip is TestCancelWhilePublishing. It runs eight goroutines that subscribe and cancel in a tight loop while another publishes twenty thousand messages. Without the lock discipline described above, that test panics with send on closed channel or trips the race detector within a few iterations, and with it, it passes. A test like that is how you find out whether a concurrency design works, and it is the one worth running with -count=20.

Ordering and what the broker guarantees

It is worth stating the guarantees precisely, because subscribers will rely on whatever the code happens to do. Messages from a single publishing goroutine arrive at a single subscriber in the order they were published, because Publish sends to each channel sequentially and a channel is a FIFO queue. Two subscribers each see the same relative order for messages from that publisher. What the broker does not promise is an order across publishers: if two goroutines publish at the same time, subscribers may see their messages interleaved in any order, and two subscribers might even see a different interleaving, since the sends to each channel happen at slightly different moments. If subscribers need one global order, the publishing has to go through a single goroutine first.

There is also no promise of delivery. A message that arrives when a buffer is full is dropped for that subscriber, and a message published before a subscriber joined is never seen by it. For many uses (cache invalidation, refreshing a UI, metrics) that is the right contract, because the next message corrects the last. For others (an audit trail, billing events) it is wrong, and the honest response is to say so in the doc comment and use something durable.

Choosing a buffer size

A buffer exists to absorb bursts, not to hide a consumer that is permanently slower than the producer. The arithmetic is short. If a publisher bursts 500 messages in 10 ms and a subscriber handles a message in 100 microseconds, the subscriber needs 50 ms to clear the burst, during which about 400 messages are waiting, so a buffer of 512 absorbs the burst and a buffer of 16 loses most of it. If the subscriber's per-message time is longer than the publisher's average gap between messages, no buffer will ever be enough, and the drop counter will grow without limit. That is the signal to fix the consumer or change the policy, not to increase the number.

I start at a small power of two (16 or 64), watch the drop counter in a staging environment under realistic bursts, and raise it only when drops come from bursts and not from sustained overload. A very large buffer is a memory bill with no bound that you will meet during an incident.

Using it from an HTTP handler

The natural client of this broker in a web service is a long-lived streaming endpoint, where each connected browser is one subscriber. The handler subscribes on entry, forwards messages as server-sent events, and cancels when the client goes away, which the request context reports:

func (s *Server) events(w http.ResponseWriter, r *http.Request) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/event-stream")

	sub := s.broker.Subscribe("orders", 32)
	defer sub.Cancel() // runs when the client disconnects or the handler returns

	for {
		select {
		case msg, ok := <-sub.C:
			if !ok {
				return // broker closed during shutdown
			}
			fmt.Fprintf(w, "data: %s\n\n", msg)
			flusher.Flush()
		case <-r.Context().Done():
			return
		}
	}
}

Three of the broker's design choices pay off here. A slow phone on a bad connection falls behind, fills its own buffer and starts dropping, without slowing the order service that publishes or the other browsers. Cancel on a closed connection returns the subscription's resources instead of leaking a goroutine and a channel per dead client. And when the process shuts down, Close ends every channel, which makes every handler's ok false and returns them all, so the HTTP server's graceful shutdown has nothing left to wait for (graceful shutdown). The details of flushing and streaming in handlers are the subject of wrapping a ResponseWriter without breaking Flush.

What I would add next, and what I would not

For production use I would add a done channel per subscription to replace the recover in PublishWait, and a Prometheus counter on drops. Wildcard topics (orders.*) are a natural request and a trap: they turn a map lookup into a pattern match on every publish, so build them only when someone needs them. Replay of recent messages for late subscribers is another common ask, and it belongs behind a separate type that keeps a ring buffer.

What I would not do is stretch this into a message queue. There is no persistence, no acknowledgement, no retry and no way across processes, and the moment one of those becomes a requirement, the answer is a real broker (NATS, Redis streams, Kafka, a cloud queue) with its Go client, not a bigger version of this one. The same goes for work distribution: if each message should go to exactly one of several consumers instead of all of them, that is a worker pool, and a plain shared channel does it.

The Go blog's concurrency patterns talk has the original channel idioms this builds on, and the sync/atomic documentation covers the typed counter used for drop counts.