Akka Stream启动后无法持续处理Elasticsearch新增数据

算力阿峰

2026-8-29

问题根因

Alpakka 自带的ElasticsearchSource默认是一次性查询实现：流启动后只会执行一次查询拉取启动时刻匹配条件的文档，拉取完成后流就结束，不会自动持续监听/轮询索引拉取后续新写入的数据，这是你改成match_all依然拿不到新数据的核心原因。
你的流配置写法完全不生效：Akka Streams 的所有操作符都是不可变设计，调用buffer、backpressureTimeout这类方法会返回带配置的新Source实例，你直接调用方法没有接收返回值，后续运行的还是原始未配置的Source，这两个参数根本没有作用到实际运行的流上。
初始版本的时间范围查询存在硬编码问题：PAST_HOUR是类加载时就计算完成的固定时间戳，就算后续做轮询，这个值也不会自动更新，永远只能查到流启动时刻往前一小时的历史数据，匹配不到后续新写入的、时间戳更大的文档。

修复方案

要实现持续消费ES新写入数据的效果，按以下步骤改造：

修正流操作的写法，所有操作符调用后保留返回的新Source实例，不要丢弃返回值导致配置失效。
增加轮询触发逻辑，定期重新执行ES查询，同时记录消费偏移量（比如最新消费到的文档时间戳、文档ID），每次查询只拉取偏移量之后的新文档，配合排序规则避免重复消费历史数据。

核心改造代码参考：

// 记录消费偏移，初始值为流启动时刻的时间戳
AtomicLong lastConsumeTs = new AtomicLong(System.currentTimeMillis());

Source<ReadResult<Stats>, NotUsed> continuousEsSource = Source
        .repeat(NotUsed.getInstance())
        // 配置轮询间隔，示例为每5秒拉取一次新数据，可根据实时性要求调整
        .throttle(1, Duration.ofSeconds(5))
        // 每次轮询动态生成查询语句，用最新偏移量做过滤
        .flatMapConcat(signal -> {
            long startTs = lastConsumeTs.get();
            // 查询语句增加排序规则，按时间戳正序拉取，避免漏数
            String dynamicQuery = "{ \"bool\": {  \"must\" : [{\"range\" : {"+
                    "\"timestamp\" : { "+
                    "\"gt\" : "+ startTs
                    +" }} }], \"sort\": [{\"timestamp\": \"asc\"}] }} ";
            return ElasticsearchSource.typed(
                    constructElasticsearchParams("data", "_doc", ApiVersion.V7),
                    dynamicQuery,
                    sourceSettings,
                    Stats.class
            );
        })
        // 消费到新记录后更新偏移量
        .map(record -> {
            // 替换为你Stats实体类中获取时间戳的实际方法
            long currentDocTs = record.source().getTimestamp();
            lastConsumeTs.accumulateAndGet(currentDocTs, Math::max);
            return record;
        })
        // 配置流参数，注意要链式调用接住返回的流实例
        .buffer(10000, OverflowStrategy.backpressure())
        .backpressureTimeout(Duration.ofSeconds(1));

// 启动改造后的持续流
continuousEsSource
        .log("es-consume-log")
        .runWith(Sink.foreach(System.out::println), actorSystem);

补充说明：这种轮询方案的实时性受轮询间隔限制，如果需要毫秒级近实时消费，不建议直接轮询ES索引，优先使用ES的CDC能力将数据变更同步到消息队列后再消费，性能和稳定性都更好。

内容的提问来源于stack exchange，提问作者blue-sky