Pub/Sub with buffered channels

Before the code: messaging and replication, in plain English

Picture a group chat. One person publishes a message. The chat app knows who is in the room and forwards the message to each of them. That is publish and subscribe, and it is the simplest piece of Redis that is not a key-value store. Replication is the same idea pointed at a backup machine: the leader sends every write to a follower, who quietly replays them. Same plumbing, different audience. In Go we wire this with buffered channels and goroutines.

Fan-out and follow-the-leader

A publisher writes once. The server forwards to every subscriber. A replica is just another subscriber, listening to the full write log.

Quiz: Quiz

Loading practice…

Go channels are made for this. Each subscriber owns a buffered chan []byte. PUBLISH iterates subscribers and does non-blocking sends. A separate goroutine drains the channel to the network. Backpressure falls out naturally.

07-pubsub/main.go
go
type Subscriber struct {
    conn net.Conn
    out  chan []byte
}

func handleConn(conn net.Conn) {
    defer conn.Close()
    sub := &Subscriber{conn: conn, out: make(chan []byte, 256)}
    go writePump(sub)
    defer func() {
        pubsub.UnsubscribeAll(sub)
        close(sub.out)
    }()
    // ... read loop parses SUBSCRIBE / PUBLISH frames
}

Every connection gets a Subscriber whose out channel buffers up to 256 pending messages. That capacity is the backpressure budget: once it fills, publishes to this subscriber start dropping.

07-pubsub/main.go
go
func (p *PubSub) Publish(ch string, message string) int {
    p.mu.RLock()
    set := p.subs[ch]
    var subs []*Subscriber
    for s := range set { subs = append(subs, s) }
    p.mu.RUnlock()

    payload := encodeMessageArray(ch, message)
    delivered := 0
    for _, s := range subs {
        select {
        case s.out <- payload:
            delivered++
        default:
            // Channel full, drop. Production would disconnect slow consumers.
        }
    }
    return delivered
}

Non-blocking send via select with default. Subscribers that fall behind get dropped messages, not blocked publishers.

07-pubsub/main.go
go
func writePump(sub *Subscriber) {
    for msg := range sub.out {
        if _, err := sub.conn.Write(msg); err != nil { return }
    }
}

One write pump goroutine per subscriber. The pump owns the conn.Write calls. Publishers never touch the socket directly.

Why a write pump? Because a slow subscriber's TCP buffer fills up, and conn.Write blocks. Without the pump, PUBLISH would block on the first slow subscriber, starving every other subscriber. The pump decouples them.

Quiz: Quiz

Loading practice…