如何在Akka Streams中调度Kafka消费者仅在特定时段运行？

Akka Streams 本身没有内置的定时启停 Source 的 API，但有两种成熟的实现模式可以满足你的需求：

方案一：扩展 RestartSource 实现时间窗口控制 #

利用你已有的 RestartSource 逻辑，在每次尝试启动消费者前检查当前时间是否在允许的窗口内，非窗口时段则延迟到下一次窗口启动，同时在流运行中加入时间检查，超出窗口时主动终止流，让 RestartSource 自动等待下一次窗口到来。

示例代码：

RestartSource.withBackoff(consumerResetProps(),
        () -> {
            LocalTime now = LocalTime.now(ZoneId.systemDefault());
            LocalTime windowStart = LocalTime.of(1, 0);
            LocalTime windowEnd = LocalTime.of(4, 0);
            boolean isInWindow = now.isAfter(windowStart) && now.isBefore(windowEnd);

            if (!isInWindow) {
                // 计算距离下一次窗口开始的延迟时间
                LocalDateTime nextWindowStart = LocalDateTime.now()
                        .with(windowStart)
                        .plusDays(now.isAfter(windowStart) ? 1 : 0);
                Duration delay = Duration.between(LocalDateTime.now(), nextWindowStart);
                
                // 延迟到窗口时间再启动消费者
                return Source.single(())
                        .delay(delay, DelayOverflowStrategy.backpressure())
                        .flatMapConcat(ignored -> 
                            Consumer.committablePartitionedSource(consumerProps(), Subscriptions.topics(topics))
                                .mapAsyncUnordered(parallelism, pair -> pair.second()
                                    .via(flow())
                                    .takeWhile(elem -> {
                                        // 处理每个元素前再次检查时间，超出窗口终止子流
                                        LocalTime current = LocalTime.now(ZoneId.systemDefault());
                                        return current.isAfter(windowStart) && current.isBefore(windowEnd);
                                    })
                                    .runWith(Committer.sink(commiterProps()), system))
                        );
            } else {
                // 窗口内正常启动，同时加入时间检查
                return Consumer.committablePartitionedSource(consumerProps(), Subscriptions.topics(topics))
                        .mapAsyncUnordered(parallelism, pair -> pair.second()
                            .via(flow())
                            .takeWhile(elem -> {
                                LocalTime current = LocalTime.now(ZoneId.systemDefault());
                                return current.isAfter(windowStart) && current.isBefore(windowEnd);
                            })
                            .runWith(Committer.sink(commiterProps()), system));
            }
        })
.toMat(Sink.ignore(), Keep.both())
.run(system);

方案二：用 Akka Scheduler 定时启停整个流 #

通过 Akka 调度器直接控制流的启动和关闭，这种方式更直观，完全符合“非窗口时段关闭消费者”的需求，且能保证优雅终止。

示例代码：

// 用于保存流的终止开关
AtomicReference<UniqueKillSwitch> killSwitchHolder = new AtomicReference<>();

// 启动消费者的逻辑
Runnable startConsumer = () -> {
    Pair<UniqueKillSwitch, CompletionStage<Done>> matValue = RestartSource.withBackoff(consumerResetProps(),
            () -> Consumer.committablePartitionedSource(consumerProps(), Subscriptions.topics(topics))
                    .mapAsyncUnordered(parallelism, pair -> pair.second()
                            .via(flow())
                            .runWith(Committer.sink(commiterProps()), system)))
            .viaMat(KillSwitches.single(), Keep.both())
            .toMat(Sink.ignore(), Keep.both())
            .run(system);
    killSwitchHolder.set(matValue.first());
};

// 关闭消费者的逻辑
Runnable stopConsumer = () -> {
    Optional.ofNullable(killSwitchHolder.get()).ifPresent(KillSwitch::shutdown);
};

// 调度每天凌晨1点启动消费者
LocalDateTime firstStart = LocalDateTime.now(ZoneId.systemDefault())
        .with(LocalTime.of(1, 0))
        .plusDays(LocalTime.now().isAfter(LocalTime.of(1, 0)) ? 1 : 0);
Duration initialStartDelay = Duration.between(LocalDateTime.now(), firstStart);
system.scheduler().scheduleAtFixedRate(
        initialStartDelay,
        Duration.ofDays(1),
        startConsumer,
        system.dispatcher()
);

// 调度每天凌晨4点关闭消费者
LocalDateTime firstStop = LocalDateTime.now(ZoneId.systemDefault())
        .with(LocalTime.of(4, 0))
        .plusDays(LocalTime.now().isAfter(LocalTime.of(4, 0)) ? 1 : 0);
Duration initialStopDelay = Duration.between(LocalDateTime.now(), firstStop);
system.scheduler().scheduleAtFixedRate(
        initialStopDelay,
        Duration.ofDays(1),
        stopConsumer,
        system.dispatcher()
);

关键注意事项 #

• 时区处理：所有时间判断必须指定时区（如 ZoneId.systemDefault()），避免因时区偏差导致时间窗口错误。
• 优雅关闭：使用 KillSwitch 或流的正常终止逻辑，确保 Kafka 偏移量正确提交，避免重复消费。
• 资源控制：方案一中需合理设置 RestartSource 的退避参数，避免非窗口时段频繁重试浪费系统资源。

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