java

10.3 实时面板数据流

eventd v3 实时面板的 WebFlux SSE 推送实现：Redis 订阅与 Lettuce 响应式驱动的接入方式、SSE 事件格式与断线重连、慢消费者下的背压与丢弃策略，以及前端 EventSource 的消费示例、命名事件监听和常见踩坑。

Leeting Yan 4 分钟阅读 1847 字

推送技术选择

eventd v3 的面板需要从服务端向浏览器推送数据。三种技术方案：

技术协议优点缺点
SSEHTTP基于 HTTP、自动重连、轻量单向推送（服务器→客户端）
WebSocketws/wss双向、低延迟需要额外协议升级
轮询HTTP简单浪费连接、延迟高

eventd v3 用 SSE（Server-Sent Events），因为：

  1. 面板只需要服务器推送，不需要客户端发数据
  2. 基于 HTTP，防火墙、代理兼容性好
  3. 浏览器原生支持 EventSource，自动重连

SSE 协议

SSE 是简单的 HTTP 协议：

GET /api/stream/metrics HTTP/1.1
Accept: text/event-stream

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

event: metric
data: {"type":"order-count","userId":"alice","count":12}
id: 1700004600000

retry: 5000

event: alert
data: {"level":"P2","message":"延迟过高"}

浏览器用 EventSource 消费：

const source = new EventSource('/api/stream/metrics');

source.addEventListener('metric', function(e) {
    const data = JSON.parse(e.data);
    updateChart(data.type, data);
});

source.addEventListener('alert', function(e) {
    const data = JSON.parse(e.data);
    showAlert(data.level, data.message);
});

source.onerror = function(e) {
    console.error('SSE error:', e);
    // 浏览器会自动重连（按 retry 参数）
};

WebFlux SSE 控制器

@RestController
public class DashboardController {

    private final ReactiveRedisTemplate<String, String> redisTemplate;

    public DashboardController(ReactiveRedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    // SSE 推送实时指标
    @GetMapping(value = "/api/stream/metrics", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> streamMetrics() {
        // 1. 先推送当前活跃指标（快照）
        Flux<ServerSentEvent<String>> snapshot = redisTemplate.opsForValue()
            .multiGet(List.of(
                "eventd:metrics:order-count:latest",
                "eventd:metrics:avg-latency:latest",
                "eventd:metrics:hot-product:latest"
            ))
            // Redis MGET 对不存在的 key 返回 null 元素。
            // 必须在交给 Flux.fromIterable 之前从 List 里剔掉：
            // fromIterable 在拉取阶段遇到 null 就抛 NPE，下游 filter 来不及生效。
            .flatMapMany(list -> Flux.fromIterable(
                list.stream().filter(Objects::nonNull).toList()))
            .map(value -> ServerSentEvent.builder(value)
                .event("snapshot")
                .build())
            // 快照失败只影响快照，不要让 onErrorResume 终止后面的 live 订阅
            .onErrorResume(e -> Flux.just(ServerSentEvent.builder(" snapshot unavailable")
                .event("error")
                .build()));

        // 2. 然后持续订阅 Redis Pub/Sub
        // 注意：ReactiveRedisConnection 没有 listenToChannel；
        // 订阅入口是 ReactiveRedisTemplate.listenTo(Topic...)
        Flux<ServerSentEvent<String>> live = redisTemplate
            .listenTo(ChannelTopic.of("eventd:metrics"))
            // sample 只保留每个时间窗内的最后一条：面板要的是「最新值」，
            // 中间值丢掉没有损失，消息再快也不会在服务端无限堆积
            .sample(Duration.ofMillis(200))
            .map(msg -> ServerSentEvent.builder(msg.getMessage())
                .event("metric")
                .build());

        return Flux.concat(snapshot, live)
            .onErrorResume(e -> {
                // 实时订阅本身出错时发送结束事件，重试逻辑在客户端。
                // 快照错误已在上面单独处理，不会走到这里。
                return Flux.just(ServerSentEvent.builder(" reconnect")
                    .event("error")
                    .build());
            });
    }

    // SSE 推送告警
    @GetMapping(value = "/api/stream/alerts", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> streamAlerts() {
        // 告警不能像指标那样 sample——采样会丢告警。
        // 这里靠告警本身的低频率；真要限流应在生产端合并同类告警
        return redisTemplate
            .listenTo(ChannelTopic.of("eventd:alerts"))
            .map(msg -> ServerSentEvent.builder(msg.getMessage())
                .event("alert")
                .build());
    }
}

背压适配

SSE 的潜在背压场景：

Flink 生产指标速度 > 浏览器消费速度
    → 服务端内存缓存增长
    → OOM

eventd v3 的背压控制策略：

  1. 按时间采样：sample(Duration) 每个时间窗只保留最后一条，消息再快也不会堆积。注意 delayElements 不是背压控制——它把元素放进内部队列逐个延后发出，下游消费慢时队列照样增长，反而放大内存问题。
  2. 丢弃旧数据：Redis 中的 latest 值只保留最新快照，不累积历史
  3. 客户端采样：前端展示每 500ms 刷新一次，不渲染每条事件
// 背压适配：高频率指标只推送最新值
public Flux<ServerSentEvent<String>> streamMetricsWithBackpressure() {
    return redisTemplate
        .listenTo(ChannelTopic.of("eventd:metrics"))
        .sample(Duration.ofMillis(500))  // 每 500ms 取最新一条
        .map(msg -> ServerSentEvent.builder(msg.getMessage())
            .event("metric")
            .build());
}

Lettuce 响应式 Redis

eventd v3 使用 Lettuce（响应式 Redis 驱动）。面板的 WebFlux 层用 Spring Data Redis 的 ReactiveRedisTemplate 包了一层，所以依赖是 spring-boot-starter-data-redis（默认实现就是 Lettuce）；下面的 XML 是只引裸驱动时的写法：

<dependency>
    <groupId>io.lettuce</groupId>
    <artifactId>lettuce-core</artifactId>
    <version>6.5.1.RELEASE</version>
</dependency>

Lettuce 的响应式 API：

// Pub/Sub 要用 connectPubSub() 的连接；普通 connect() 拿到的 reactive()
// 里没有 subscribe/observeChannels
RedisClient client = RedisClient.create("redis://localhost:6379");
StatefulRedisPubSubConnection<String, String> pubSub = client.connectPubSub();
RedisPubSubReactiveCommands<String, String> reactive = pubSub.reactive();

// 消息流来自 observeChannels()；subscribe() 只是「发起订阅」，返回 Mono<Void>，
// 不携带任何消息。先建立观察流再订阅，避免漏掉最初几条消息
reactive.observeChannels()
    .filter(msg -> "eventd:metrics".equals(msg.getChannel()))
    .doOnNext(msg -> System.out.println("Received: " + msg.getMessage()))
    .subscribe();
reactive.subscribe("eventd:metrics").subscribe();

面板这一层用 Spring Data Redis 的 ReactiveRedisTemplate.listenTo(ChannelTopic.of("eventd:metrics")) 更省事，订阅/退订/取消都由它托管；上面这段是为了说明 Lettuce 原始 API 的分工。

面板前端

<!DOCTYPE html>
<html>
<head>
    <title>eventd v3 实时面板</title>
</head>
<body>
    <div id="metrics">
        <div id="qps">QPS: --</div>
        <div id="latency">延迟: --</div>
        <div id="errors">错误率: --</div>
    </div>
    <div id="alerts"></div>

    <script>
        // 指标流。服务端用的是命名事件（event: metric / event: snapshot），
        // onmessage 只接收没有 event 字段的消息，命名事件必须用 addEventListener，
        // 写成 onmessage 会一条都收不到
        const metricSource = new EventSource('/api/stream/metrics');
        function renderMetric(e) {
            const data = JSON.parse(e.data);
            document.getElementById('qps').textContent = 'QPS: ' + data.qps;
            document.getElementById('latency').textContent =
                '延迟 P99: ' + data.latencyP99 + 'ms';
        }
        metricSource.addEventListener('metric', renderMetric);
        metricSource.addEventListener('snapshot', renderMetric);

        // 告警流
        const alertSource = new EventSource('/api/stream/alerts');
        alertSource.addEventListener('alert', function(e) {
            const data = JSON.parse(e.data);
            const div = document.createElement('div');
            div.className = 'alert alert-' + data.level.toLowerCase();
            div.textContent = `[${data.level}] ${data.message}`;
            document.getElementById('alerts').appendChild(div);
            // 5 秒后自动移除
            setTimeout(() => div.remove(), 5000);
        });
    </script>
</body>
</html>

部署架构

Kafka Cluster
    ├── eventd-raw
    └── eventd-cleaned
           ↓
    Flink JobManager + TaskManager (eventd-analytics)
           ↓
    Redis Cluster
        ├── eventd:metrics:* (定时更新)
        └── eventd:alerts (实时发布)
           ↓
    WebFlux Server (eventd-panel)  × N (负载均衡)
           ↓
    浏览器 (EventSource SSE)

可观测性

// 面板服务的监控端点
@GetMapping("/api/health")
public Map<String, Object> health() {
    Map<String, Object> health = new HashMap<>();
    health.put("sseConnections", sseConnectionCount.get());
    // isOpen() 在 StatefulRedisConnection 上，RedisClient 本身没有这个方法
    health.put("redisConnected", redisConnection.isOpen());
    health.put("timestamp", Instant.now());
    return health;
}

与卷二的衔接

eventd v3 的面板基于卷二的技术：

卷二内容v3 用法
WebFluxSSE 控制器
ReactorRedis 响应式驱动
R2DBC不直接使用（v3 用 Redis 而非数据库）

小结

组件技术职责
SSEHTTP + EventSource服务器→浏览器推送
Lettuce响应式 Redis订阅 Redis Pub/Sub
WebFlux@GetMapping(produces = TEXT_EVENT_STREAM)响应式 SSE 端点
背压sample 按时间窗取最新防止客户端过载
面板HTML + JS实时展示

第 10 章结束。下一节进入端到端一致性：11.1 端到端 Exactly-Once 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南