Server-Sent Events Across Multiple Pods: Redis Pub/Sub + Spring WebFlux

This article was originally published on Jo4 Blog.

We replaced our notification-bell polling with Server-Sent Events. On a single pod, this is easy — Spring WebFlux + a Flux of events, done in 30 lines. On multiple pods, it gets interesting fast: a write that happens on pod A needs to reach a subscriber on pod B, and naively, it doesn't.

Here's the full pattern: how the SSE endpoint is shaped, how Redis pub/sub fans events across pods, and the subtle ordering and lifecycle gotchas we hit along the way.

The Endpoint

A /notifications/stream SSE endpoint that emits the current unread count on subscribe, then live updates whenever the count changes:

@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public ResponseEntity<Flux<ServerSentEvent<NotificationStreamEvent>>>
        streamNotifications() {
    authContext.requireScope("read");
    Long userId = authContext.getCurrentUser().getId();

    ServerSentEvent<NotificationStreamEvent> initial = ServerSentEvent
        .<NotificationStreamEvent>builder()
        .event("unread-count")
        .data(NotificationStreamEvent.unreadCount(
            notificationService.getUnreadCount(userId)))
        .build();

    Flux<ServerSentEvent<NotificationStreamEvent>> events =
        notificationEventBus.subscribe(userId)
            .map(e -> ServerSentEvent.<NotificationStreamEvent>builder()
                .event(e.type()).data(e).build());

    Flux<ServerSentEvent<NotificationStreamEvent>> heartbeat =
        Flux.interval(Duration.ofSeconds(30))
            .map(t -> ServerSentEvent.<NotificationStreamEvent>builder()
                .comment("ping").build());

    Flux<ServerSentEvent<NotificationStreamEvent>> body = Flux.concat(
            Flux.just(initial),
            Flux.merge(events, heartbeat))
        .take(Duration.ofMinutes(30));

    HttpHeaders headers = apiHeaders();
    headers.set(HttpHeaders.CACHE_CONTROL, "no-cache");
    headers.set("X-Accel-Buffering", "no");
    return ResponseEntity.ok().headers(headers).body(body);
}

A few things worth highlighting:

Initial emit on subscribe. The first ServerSentEvent carries the current count synchronously. Clients get the answer immediately and don't need a separate REST round-trip for first paint.

Heartbeat every 30 seconds. SSE looks idle to TCP intermediaries. Comment events (:ping) keep the connection alive through any proxy/CDN idle timeout.

X-Accel-Buffering: no. Reverse proxies (nginx and friends) buffer responses by default. SSE is point-the-other-way — buffering means events sit in the proxy until the buffer fills, defeating the entire stream. This header is a hint to skip buffering for this response.

take(Duration.ofMinutes(30)). Server-initiated close. The stream lives for 30 minutes, then the server hangs up. The client reconnects with a fresh JWT. This is how we handle mid-stream token expiry without inventing our own re-validation logic — close-and-reconnect is the mechanism.

The Per-User Sink

The notificationEventBus.subscribe(userId) call lazily creates a per-user Sinks.Many<NotificationStreamEvent> and returns its Flux. New subscribers latch onto the same sink; the sink is removed when the last subscriber unsubscribes:

public Flux<NotificationStreamEvent> subscribe(Long userId) {
    return Flux.defer(() -> {
        Sinks.Many<NotificationStreamEvent> sink = sinks.computeIfAbsent(userId,
            k -> Sinks.many().multicast().directBestEffort());
        return sink.asFlux();
    }).doFinally(signal -> cleanupIfEmpty(userId));
}

private void cleanupIfEmpty(Long userId) {
    sinks.computeIfPresent(userId, (k, sink) ->
        sink.currentSubscriberCount() == 0 ? null : sink);
}

Two non-obvious choices:

Flux.defer instead of Flux.create or eager construction. Without defer, this race fires:

1. subscribe(userId) returns a Flux referencing a sink.
2. Before the downstream attaches, the previous subscriber for that user unsubscribes.
3. cleanupIfEmpty removes the sink from the map.
4. A publish arrives, looks up the sink in the map, finds nothing, drops the event.
5. The downstream finally attaches — to the now-orphaned sink.

Flux.defer rebuilds the lookup at the moment downstream attaches, which makes the computeIfAbsent/attach pair effectively atomic from the publisher's perspective.

directBestEffort backpressure. A slow subscriber drops events rather than terminating the stream. For unread-count specifically, this is correct: the latest count replaces any dropped one, so dropping is harmless.

The Cross-Pod Problem

So far so good on one pod. The break comes the moment you scale horizontally.

Pod A holds the SSE connection for User 42. A write on pod B increments User 42's unread count and publishes to pod B's local sink map. Pod A's sink map has no idea anything happened. The user sees nothing change.

The fix is Redis pub/sub. Every pod subscribes to a Redis pattern; every write fans out to that pattern; every pod with a local subscriber for the affected user emits to its local sink.

@Component
public class RedisNotificationEventBus implements NotificationEventBus, MessageListener {

    public static final String CHANNEL_PREFIX = "jo4:notify:user:";
    public static final String CHANNEL_PATTERN = CHANNEL_PREFIX + "*";

    private final ConcurrentHashMap<Long, Sinks.Many<NotificationStreamEvent>> sinks
        = new ConcurrentHashMap<>();

    @Override
    public void publish(Long userId, NotificationStreamEvent event) {
        String channel = CHANNEL_PREFIX + userId;
        try {
            String payload = objectMapper.writeValueAsString(event);
            redisTemplate.convertAndSend(channel, payload);
        } catch (Exception e) {
            log.warn("Redis publish failed for userId={}; falling back to local emit",
                userId, e);
            emitLocally(userId, event);
        }
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            String channel = new String(message.getChannel(), StandardCharsets.UTF_8);
            if (!channel.startsWith(CHANNEL_PREFIX)) return;
            Long userId = Long.parseLong(channel.substring(CHANNEL_PREFIX.length()));
            NotificationStreamEvent event = objectMapper.readValue(
                message.getBody(), NotificationStreamEvent.class);
            emitLocally(userId, event);
        } catch (Exception e) {
            // Never throw from MessageListener — would tear down the listener container.
            log.error("Failed to handle pub/sub on channel={}", message.getChannel(), e);
        }
    }
}

The trick:

• Publishes go to Redis only. They don't touch the local sinks directly. Even pod A's own writes route through Redis, which means every replica's local sinks get updated through one consistent path (no "but we wrote it locally" branch to maintain).
• Subscribes happen on every pod, via MessageListener. The pattern subscription is wired in a @Configuration class (MessageListenerContainer.addMessageListener(bus, new PatternTopic("jo4:notify:user:*"))).
• Routing is by channel suffix. Channel jo4:notify:user:42 carries events for user 42; every pod that has a local sink for 42 emits, every pod that doesn't ignores the message. No global broadcast to every subscriber.

The Fallback to Local

Notice this branch:

} catch (Exception e) {
    log.warn("Redis publish failed for userId={}; falling back to local emit",
        userId, e);
    emitLocally(userId, event);
}

Redis pub/sub is fire-and-forget and the connection can die. If the publish fails, we still emit to the local sink so users on the current pod don't miss the event. Users on other pods miss it during the outage, but that's strictly better than everyone missing it.

The Transactional-Listener Bridge

How do events actually get into bus.publish(...)? Not from the controller — the controller is a passive subscriber. They come from the rest of the application, via Spring's transactional event listener:

@Component
@RequiredArgsConstructor
public class UnreadCountStreamPublisher {

    private final NotificationService notificationService;
    private final UnreadCountCacheService cacheService;
    private final NotificationEventBus eventBus;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onUnreadCountChanged(UnreadCountChangedEvent event) {
        Long userId = event.userId();
        try {
            long count = notificationService.getUnreadCountFresh(userId);
            cacheService.evict(userId);
            eventBus.publish(userId, NotificationStreamEvent.unreadCount(count));
        } catch (Exception e) {
            // Listener must not throw — would mark the just-committed transaction
            // as failed in some configurations and disrupt unrelated handlers.
            log.error("Failed to publish unread-count change for userId={}",
                userId, e);
        }
    }
}

Three things this is doing right:

AFTER_COMMIT, not BEFORE_COMMIT. The SSE event must not fire until the DB write commits. Otherwise the client could see a count that's about to be rolled back.

Fresh DB read, not the cached value. After a write, the cache may briefly hold the pre-write count. Reading fresh from the DB is the only safe authority for what we're publishing. Then we evict the cache (rather than write-through it) so the next reader hits the DB and re-populates.

Catch-all that never throws. A @TransactionalEventListener that throws from AFTER_COMMIT can mark the just-committed transaction as failed in some configurations. Listeners must be defensive.

The application code that mutates unread counts publishes a UnreadCountChangedEvent and moves on. It doesn't know about Redis, doesn't know about SSE, doesn't know about cache eviction. All of that is downstream of the event.

Lessons Learned

• A single-pod SSE stream is simple. A multi-pod SSE stream with shared subscribers needs a fan-out plane. Redis pub/sub is the cheapest one that works.
• Always route publishes through the same path, even local ones. "Publish locally + publish to Redis" creates double-delivery possibilities that "publish to Redis only, subscribe from everywhere" avoids by construction.
• Backpressure is a product decision. For unread counts, dropping is correct because newer values supersede older ones. For chat messages, dropping is incorrect. Pick directBestEffort, buffer, or failFast per use case.
• Flux.defer for any sink that may be created/destroyed by lifecycle. Eager Flux construction races against unsubscribe-then-subscribe sequences in subtle ways.
• Server-initiated close is the right way to handle JWT expiry on long-lived streams. Hang up after a fixed lifetime; the client reconnects with a fresh token. No mid-stream auth gymnastics.
• MessageListener callbacks must never throw. A thrown exception tears down the listener container and your app stops receiving events with no obvious symptom.
• AFTER_COMMIT is the only safe phase for "broadcast that the data changed" listeners. Anything earlier risks broadcasting a value that's about to roll back.

---

How did you scale your real-time stream beyond one process? Pub/sub, sticky sessions, websocket cluster? Drop your pattern in the comments.

Building jo4.io — a URL shortener with real-time updates that work the same on one pod or twenty.