Spring Cloud Stream：优雅停止响应式Supplier并确保数据推送完成

你的问题核心在于：doOnComplete触发时仅表示上游Flux的数据流已发送完毕，但Kinesis绑定器可能仍在异步处理剩余消息的推送，直接关闭应用会导致部分消息丢失。手动sleep是治标不治本的方法，以下是合理的解决方案：

核心问题点 #

你忽略了响应式流完成信号与绑定器消息发送完成的异步性：Flux的完成事件只是上游数据生产结束的标记，而绑定器（如Kinesis）通常会在后台异步处理消息发送，两者的完成时机并不一致。

解决方案 #

1. 利用Spring Cloud Stream绑定生命周期管理 #

不要手动调用System.exit()或直接关闭上下文，而是依赖Spring Cloud Stream的绑定生命周期钩子，等待所有绑定的消息发送操作完成后再关闭应用：

@Autowired
private ApplicationContext applicationContext;

public Supplier<Flux<Map<String, List<Object>>>> streamingDataSource() {
    Sinks.Many<Map<String, List<Object>>> sink = Sinks.many()
            .unicast()
            .onBackpressureBuffer();

    scheduler.schedule(() -> source.accept(sink));

    return () -> sink.asFlux()
            .subscribeOn(Schedulers.boundedElastic())
            .doFinally(signalType -> {
                // 获取BindingService并停止所有生产者绑定
                BindingService bindingService = applicationContext.getBean(BindingService.class);
                for (Binding binding : bindingService.getBindings()) {
                    if (binding instanceof MessageProducer) {
                        ((MessageProducer) binding).stop();
                    }
                }
                // 优雅关闭Spring上下文
                applicationContext.close();
            })
            .log();
}

2. 跟踪消息发送的异步结果 #

如果需要更精准地确认每个消息都已发送成功，可以通过自定义逻辑收集所有消息的发送结果，等待全部完成后再触发关闭：

public Supplier<Flux<Map<String, List<Object>>>> streamingDataSource() {
    Sinks.Many<Map<String, List<Object>>> sink = Sinks.many()
            .unicast()
            .onBackpressureBuffer();

    scheduler.schedule(() -> source.accept(sink));

    // 用于跟踪所有消息发送结果的线程安全集合
    List<CompletableFuture<Void>> sendFutures = new CopyOnWriteArrayList<>();

    return () -> sink.asFlux()
            .subscribeOn(Schedulers.boundedElastic())
            .doOnNext(message -> {
                // 发送时记录异步结果，实际逻辑需结合绑定器调整
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    // 此处替换为绑定器的消息发送逻辑
                }, Schedulers.boundedElastic());
                sendFutures.add(future);
            })
            .doFinally(signalType -> {
                // 等待所有发送任务完成
                CompletableFuture.allOf(sendFutures.toArray(new CompletableFuture[0])).join();
                applicationContext.close();
            })
            .log();
}

3. 配置Spring优雅关闭超时 #

在配置文件中设置优雅关闭的超时时间，确保容器有足够时间处理剩余任务：

spring:
  lifecycle:
    timeout-per-shutdown-phase: 30s

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