Streaming LLM outputs to multiple downstream clients demands a concurrency primitive that neither blocks the generator nor loses tokens under load. A go channel pub sub llm stream fan-out pattern lets you multiplex a single token stream from a model endpoint to many subscribers using goroutines and buffered channels, and it fits Go’s model better than callback hell.
Why naive multiplexing fails
If you pipe one io.Reader from an LLM API directly into N websockets in a single loop, the slowest socket dictates the pace of the entire system. The HTTP client’s read buffer fills, the provider’s stream stalls, and you start hitting request timeouts. I’ve seen this cause cascading failures where a single broken browser tab took down a whole inference batch.
The fix is to decouple the producer from consumers. Channels give you that decoupling with built-in blocking semantics, and a hub goroutine can own the subscriber set to avoid lock contention on the hot path.
Define the hub structure
Start with a hub that owns a single broadcast channel and manages registration via its own channels. This go channel pub sub llm stream fan-out design avoids mutexes on the hot path, which matters when you’re pushing hundreds of tokens per second to thousands of subscribers.
type Hub struct {
register chan chan string
unregister chan chan string
broadcast chan string
subs map[chan string]struct{}
}
func NewHub() *Hub {
return &Hub{
register: make(chan chan string),
unregister: make(chan chan string),
broadcast: make(chan string, 256),
subs: make(map[chan string]struct{}),
}
}The buffer on broadcast absorbs bursts when the hub goroutine is busy registering new clients. Size it based on expected token rate times max scheduling delay; 256 is a reasonable starting point for a single model stream.
Run the hub loop
The hub’s run method is the only place that touches the subscriber map. This serializes mutations and reads, eliminating race conditions without a mutex on the hot path.
func (h *Hub) Run(ctx context.Context) {
for {
select {
case <-ctx.Done():
for ch := range h.subs {
close(ch)
}
return
case ch := <-h.register:
h.subs[ch] = struct{}{}
case ch := <-h.unregister:
if _, ok := h.subs[ch]; ok {
delete(h.subs, ch)
close(ch)
}
case msg := <-h.broadcast:
for ch := range h.subs {
select {
case ch <- msg:
default:
// subscriber is behind; drop token
}
}
}
}
}The default case in the inner select is a deliberate tradeoff. Blocking on ch <- msg would stall the hub and therefore all other subscribers. For LLM token streams, dropping a token on a hopelessly slow client is usually better than poisoning the whole fan-out. If you need guaranteed delivery, use a larger per-subscriber buffer and a separate timeout, but accept that memory grows with lagging clients.
Subscribe and unsubscribe safely
Expose a method that returns a receive-only channel and a cleanup function. The cleanup sends the channel to unregister and must be called exactly once.
func (h *Hub) Subscribe() (<-chan string, func()) {
ch := make(chan string, 64)
h.register <- ch
var once sync.Once
unsub := func() {
once.Do(func() { h.unregister <- ch })
}
return ch, unsub
}Note the per-subscriber buffer of 64 tokens. That gives a client roughly half a second of slack at 128 tokens/sec before drops begin. Tune this to your p95 consumer speed.
A common bug: calling unsub twice will send the same channel to unregister twice, causing a double close. The sync.Once guard prevents that panic. Without it, you’ll see panic: close of closed channel in production during reconnect storms.
Wiring an LLM stream into the hub
Point a standard HTTP client at an OpenAI-compatible streaming endpoint. n4n.ai exposes one such endpoint covering 240+ models with automatic fallback when a provider is degraded, so the same parsing code works across model swaps. The key is to scan the response body line by line and push only the delta tokens onto hub.broadcast.
func streamToHub(ctx context.Context, hub *Hub, payload []byte) error {
req, _ := http.NewRequestWithContext(ctx, "POST",
"https://api.n4n.ai/v1/chat/completions", bytes.NewReader(payload))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+os.Getenv("KEY"))
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
scanner := bufio.NewScanner(resp.Body)
scanner.Buffer(make([]byte, 1024*1024), 1024*1024)
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data: ") {
continue
}
data := strings.TrimPrefix(line, "data: ")
if data == "[DONE]" {
break
}
var chunk struct {
Choices []struct {
Delta struct {
Content string `json:"content"`
} `json:"delta"`
} `json:"choices"`
}
if err := json.Unmarshal([]byte(data), &chunk); err != nil {
continue
}
if len(chunk.Choices) > 0 {
token := chunk.Choices[0].Delta.Content
if token != "" {
hub.broadcast <- token
}
}
}
return scanner.Err()
}This blocks on hub.broadcast only when the hub’s internal buffer is full, which means your provider connection is faster than the hub can dispatch. In practice, the hub loop is cheap and the 256-deep buffer hides microseconds of contention.
Handling slow consumers without killing the hub
If you cannot tolerate dropped tokens, replace the default drop with a per-subscriber goroutine that uses a timeout:
case msg := <-h.broadcast:
for ch := range h.subs {
go func(c chan string, m string) {
select {
case c <- m:
case <-time.After(200 * time.Millisecond):
// log slow consumer, do not block hub
}
}(ch, msg)
}Spawning a goroutine per token per subscriber is wasteful at scale. A better architecture is to give each subscriber its own forwarding goroutine at registration time, reading from a shared broadcast channel and writing to the subscriber channel with a timeout. That moves the concurrency out of the hub loop entirely.
case ch := <-h.register:
h.subs[ch] = struct{}{}
go func(c chan string) {
for msg := range h.broadcast {
select {
case c <- msg:
case <-time.After(200 * time.Millisecond):
return
}
}
}(ch)But then broadcast must be fanned out to all those goroutines; you need a separate fan-out step. The simplest correct version: keep the hub loop as the single writer to subscriber channels, but make subscriber channels large and monitor drop metrics. Most chat UIs don’t care about a missed token when a client is already 2 seconds behind.
Graceful shutdown
On process stop, cancel the context passed to hub.Run. The loop closes all subscriber channels, which signals downstream writers (e.g., WebSocket pumps) to exit. Your WebSocket handler should range over the subscriber channel and break on channel close.
go func() {
for token := range subCh {
if err := conn.WriteMessage(websocket.TextMessage, []byte(token)); err != nil {
break
}
}
}()If you don’t close the channels, those goroutines leak. I’ve debugged memory graphs that looked like stairs purely because an unsubscribe path forgot to signal the hub.
Testing the fan-out
Write a table test that starts the hub, subscribes N times, broadcasts M tokens, and asserts each subscriber received M (or M minus expected drops). Use sync.WaitGroup and a collecting goroutine per subscriber.
func TestFanOut(t *testing.T) {
h := NewHub()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go h.Run(ctx)
ch, unsub := h.Subscribe()
defer unsub()
go func() {
for i := 0; i < 10; i++ {
h.broadcast <- "tok"
}
cancel()
}()
count := 0
for range ch {
count++
if count == 10 { break }
}
if count != 10 {
t.Fatalf("got %d", count)
}
}Tradeoffs and final notes
The go channel pub sub llm stream fan-out approach is not a distributed message queue. It lives in one process and loses all buffered tokens on restart. If you need cross-instance broadcast, put Redis pub/sub behind the hub. Also, channel counts are not free: each subscriber goroutine and buffer consumes memory, so cap concurrent streams per node.
Use buffered channels sized to your p95 lag, drop on slow clients by default, and keep the hub loop single-threaded for correctness. That gives you a predictable, low-latency multiplexer for LLM token streams without pulling in a heavier framework.