如何在Akka Streams自定义图阶段中封装Pub-Sub模式Java API

云原生阿岚

2026-10-2

适配性结论

Akka Streams 完全适配该场景，是这类需求的最优选择之一：

原生支持异步回调式数据源的对接，内置背压机制可以避免上游推数据速度快于下游处理速度时的OOM问题
内置的资源生命周期管理能力，可以自动处理订阅创建、销毁的逻辑，避免资源泄漏
丰富的流式操作符可以直接对实时事件做过滤、转换、聚合、窗口计算等各类处理，不需要自己重复造轮子

具体实现方案

不需要自行实现自定义图阶段，直接使用Akka Streams官方提供的Source.queue组件即可，该组件已经封装了线程安全、背压处理、生命周期管理的所有逻辑，实现步骤如下：

第一步：定义队列参数，根据你的业务场景选择缓冲区大小和溢出策略，对数据完整性要求高的场景选择OverflowStrategy.backpressure()，允许丢部分旧数据的场景可以选OverflowStrategy.dropHead()
第二步：将第三方API的回调逻辑和队列绑定，同时将订阅的生命周期和流的生命周期绑定，流终止时自动取消订阅释放资源
第三步：直接基于队列返回的Source编写后续的流处理逻辑

import akka.Done;
import akka.actor.ActorSystem;
import akka.stream.OverflowStrategy;
import akka.stream.QueueOfferResult;
import akka.stream.javadsl.RunnableGraph;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
import akka.stream.javadsl.SourceQueueWithComplete;
import java.util.List;
import java.util.concurrent.CompletionStage;

// 1. 创建带背压的Source队列，缓冲区大小可根据实际吞吐调整
Source<Event, SourceQueueWithComplete<Event>> eventSource = Source.queue(
    1024, // 缓冲区大小，可根据实际业务调整
    OverflowStrategy.backpressure() // 缓冲区满时向上游回调施加背压
);

// 2. 构建流运行逻辑
RunnableGraph<SourceQueueWithComplete<Event>> graph = eventSource
    // 绑定流生命周期与订阅生命周期
    .watchTermination((queue, terminationFuture) -> {
        // 创建第三方API订阅
        Subscription sub = createSubscription();
        sub.addListener(new Listener() {
            @Override
            public void eventsReceived(List<Event> events) {
                // 将事件批量推送到队列
                events.forEach(event -> {
                    CompletionStage<QueueOfferResult> offerResult = queue.offer(event);
                    offerResult.whenComplete((result, throwable) -> {
                        // 推送失败时销毁订阅避免资源泄漏
                        if (throwable != null || result instanceof QueueOfferResult.Failure) {
                            sub.cancel();
                            queue.fail(throwable != null ? throwable : ((QueueOfferResult.Failure) result).cause());
                        }
                    });
                });
            }
        });
        // 流正常终止或异常终止时都销毁订阅
        terminationFuture.whenComplete((done, throwable) -> {
            sub.cancel();
            if (throwable != null) {
                queue.fail(throwable);
            } else {
                queue.complete();
            }
        });
        return queue;
    })
    // 后续可直接接任意流处理逻辑，示例为简单打印
    .map(event -> {
        // 此处替换为你的事件转换、过滤、聚合等处理逻辑
        return event;
    })
    .to(Sink.foreach(System.out::println)); // 此处替换为你的实际输出Sink

// 3. 启动流
ActorSystem system = ActorSystem.create("event-stream-system");
SourceQueueWithComplete<Event> runQueue = graph.run(system);

注意：如果第三方API的eventsReceived回调返回的事件批次较大，可以直接批量提交到队列，不需要循环单条提交，进一步提升处理效率。

之前自定义图阶段容易失败的核心原因是，自定义GraphStage需要自行处理跨线程同步、背压信号响应、生命周期回调等多个边界逻辑，稍有不慎就会出现线程安全问题、背压失效、资源泄漏等问题，非必要不建议自行实现，优先使用官方封装好的队列组件。

内容的提问来源于stack exchange，提问作者Ethel Merman