Real-Time Data Streaming with Server-Sent Events (SSE) and Go: Concurrency & High-Throughput Pipelines

Table of Contents

Real-Time Communication: Choosing the Right Protocol

Live notification feeds, order-book tickers, and infrastructure dashboards share a common requirement: pushing data instantly to the browser. Standard HTTP polling struggles here. When 10,000 clients ping a server every second, that creates 10,000 requests per second of HTTP header parsing, TLS overhead, and database query churn.

Modern architectures typically narrow down real-time updates to three core choices:

Short/Long Polling: The client polls repeatedly or holds an open connection until new data arrives. It works without special infrastructure, but each poll wastes bandwidth on headers (often 500B to 1KB per request) and burns connection slots.
WebSocket: Upgrades standard HTTP to a persistent, bidirectional TCP connection. It is the gold standard for full-duplex workloads like multiplayer games or real-time collaborative whiteboards. However, it requires custom reconnect logic, heartbeats, and dedicated proxy configurations.
Server-Sent Events (SSE): A lightweight standard built directly on top of HTTP. The server keeps a persistent response open and streams plain UTF-8 text down to the client. Browsers handle reconnections out of the box, enterprise proxies route it cleanly, and HTTP/2 multiplexing comes for free.

When to Choose Server-Sent Events

SSE offers a sweet spot for unidirectional feeds where clients only need to consume data, not publish it over the same socket.

Key Advantages

Frictionless Network Compatibility: SSE streams over standard ports 80 and 443. Load balancers, API gateways, and corporate firewalls treat it as regular HTTP traffic without special WebSocket upgrade handshakes.
Built-In Resilience: The browser’s native EventSource API automatically reconnects if a dropped connection occurs. It even passes a Last-Event-ID header so your backend can resume the stream right where it cut off.
Tiny Memory Footprint: Go allocates roughly 2 KB of initial stack memory per goroutine. A lightweight Go server with 4 GB of RAM can comfortably manage 50,000 concurrent SSE subscribers without breaking a sweat.

Trade-offs & Constraints

Strictly Unidirectional: Data moves one way: server to client. Any upstream message from the client still requires an independent HTTP POST or PUT request.
Browser Connection Caps on HTTP/1.1: Browsers enforce a hard limit of 6 simultaneous open HTTP/1.1 connections per domain. You should serve SSE over HTTP/2, where stream multiplexing allows hundreds of streams over a single TCP connection.

Architecture: The Thread-Safe Go Broker Pattern

When thousands of clients subscribe simultaneously, concurrency control becomes your primary challenge. Naive implementations often trigger race conditions when a client disconnects while the server is broadcasting.

Writing directly to an unbuffered or closed channel in Go will cause panics and blocked goroutines. Similarly, reading and writing to a raw client map across multiple goroutines causes instant runtime crashes.

A centralized Broker pattern solves this cleanly. A single dedicated loop owns the client registry and processes subscription, unsubscription, and broadcast events sequentially through dedicated channels.

Step-by-Step Implementation

Here is how to build a resilient SSE broker using only the Go standard library.

Step 1: Implementing the Thread-Safe Broker

First, define the broker state and its event loop to manage client registrations and message fan-out safely.

package main

import (
	"fmt"
	"log"
	"net/http"
	"time"
)

type Client chan string

type Broker struct {
	clients        map[Client]bool
	newClients     chan Client
	closingClients chan Client
	notifier       chan string
}

func NewBroker() *Broker {
	return &Broker{
		clients:        make(map[Client]bool),
		newClients:     make(chan Client),
		closingClients: make(chan Client),
		notifier:       make(chan string),
	}
}

func (b *Broker) Start() {
	for {
		select {
		case s := <-b.newClients:
			b.clients[s] = true
			log.Printf("Client connected. Active clients: %d", len(b.clients))

		case s := <-b.closingClients:
			if _, ok := b.clients[s]; ok {
				delete(b.clients, s)
				close(s)
				log.Printf("Client disconnected. Active clients: %d", len(b.clients))
			}

		case event := <-b.notifier:
			for s := range b.clients {
				select {
				case s <- event:
				default:
					// Buffer full or slow client; avoid blocking the entire broker
					log.Println("Skipping slow client")
				}
			}
		}
	}
}

Step 2: Creating the HTTP SSE Handler

Next, write the HTTP handler. It validates flusher support, sets streaming headers, registers the client channel, and flushes bytes as soon as they arrive.

func (b *Broker) ServeHTTP(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")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("Access-Control-Allow-Origin", "*")

	messageChan := make(Client, 10)
	b.newClients <- messageChan

	defer func() {
		b.closingClients <- messageChan
	}()

	ctx := r.Context()

	for {
		select {
		case <-ctx.Done():
			return
		case msg, ok := <-messageChan:
			if !ok {
				return
			}
			// SSE standard format: data: <payload>\n\n
			fmt.Fprintf(w, "data: %s\n\n", msg)
			flusher.Flush()
		}
	}
}

Step 3: Emitting Live Telemetry

Spin up the broker and attach a background worker that publishes CPU metrics every two seconds.

func main() {
	broker := NewBroker()
	go broker.Start()

	// Simulate real-time data events
	go func() {
		for {
			time.Sleep(2 * time.Second)
			payload := fmt.Sprintf(`{"timestamp": %d, "metric": "cpu_load", "value": %.2f}`, 
				time.Now().Unix(), 
				float64(time.Now().UnixNano()%100)/10.0)
			broker.notifier <- payload
		}
	}()

	http.Handle("/events", broker)

	log.Println("SSE server started on :8080. Endpoint: /events")
	if err := http.ListenAndServe(":8080", nil); err != nil {
		log.Fatalf("Server failed: %v", err)
	}
}

Step 4: Testing the Stream

Test the raw event stream from your terminal using curl with the unbuffered -N flag:

curl -N -H "Accept: text/event-stream" http://localhost:8080/events

Consuming the stream in the browser takes only a few lines of standard JavaScript:

const eventSource = new EventSource('http://localhost:8080/events');

eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log('Real-time metric received:', data);
};

eventSource.onerror = (err) => {
    console.error('SSE connection error:', err);
};