如何使用Alpakka、Spring Boot及Akka-stream初始化持续运行的流

你可以借助Spring Boot的启动生命周期回调接口自动启动流，无需外部API触发，完整实现方案如下：

• 第一步：托管核心依赖到Spring容器
  将Akka ActorSystem、Alpakka Kafka消费者配置、待运行的流逻辑定义为Spring Bean，示例代码如下：

@Configuration
public class AkkaKafkaConfig {
    @Bean
    public ActorSystem actorSystem() {
        return ActorSystem.create("kafka-consumer-system");
    }

    @Bean
    public ConsumerSettings<String, String> consumerSettings(ActorSystem actorSystem) {
        return ConsumerSettings.create(actorSystem, new StringDeserializer(), new StringDeserializer())
                .withBootstrapServers("你的Kafka Broker地址")
                .withGroupId("你的消费组ID")
                .withAutoOffsetReset(AutoOffsetReset.EARLIEST);
    }

    // 定义Kafka消费流逻辑
    @Bean
    public Runnable kafkaConsumeStream(ConsumerSettings<String, String> consumerSettings, ActorSystem actorSystem) {
        return () -> Consumer.plainSource(consumerSettings, Subscriptions.topics("你要消费的Topic名称"))
                .runForeach(record -> {
                    // 此处写入你的业务处理逻辑
                    System.out.printf("接收到消息：key=%s，value=%s%n", record.key(), record.value());
                }, actorSystem)
                .whenComplete((done, throwable) -> {
                    if (throwable != null) {
                        log.error("Kafka消费流异常退出", throwable);
                    }
                });
    }
}

• 第二步：实现启动回调自动触发流运行
  实现ApplicationRunner接口，Spring Boot应用上下文初始化完成后会自动执行该接口的run方法，直接在该方法中启动定义好的流即可：

@Component
public class KafkaStreamStarter implements ApplicationRunner {
    private final Runnable kafkaConsumeStream;
    private CompletionStage<Done> streamRunningStage;

    public KafkaStreamStarter(Runnable kafkaConsumeStream) {
        this.kafkaConsumeStream = kafkaConsumeStream;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 应用启动完成自动启动Kafka消费流
        kafkaConsumeStream.run();
    }

    // 优雅停机配置：应用关闭时主动停止流、释放资源
    @PreDestroy
    public void stopStreamOnShutdown() {
        if (streamRunningStage != null) {
            streamRunningStage.toCompletableFuture().cancel(true);
        }
    }
}

• 优化项：增加流异常自动重启能力
  如果需要避免流因为临时异常终止后无法自动恢复，可以用Alpakka提供的RestartSource包裹消费逻辑，配置退避重启策略：

RestartSource.onFailuresWithBackoff(
        RestartSettings.create(Duration.ofSeconds(3), Duration.ofSeconds(30), 0.2),
        () -> Consumer.plainSource(consumerSettings, Subscriptions.topics("你要消费的Topic名称"))
).runForeach(record -> {
    // 业务处理逻辑
}, actorSystem);

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