反应式Kafka生产消息时程序提前退出，如何保障全量消息成功投递

AI 小北

2026-11-5

问题根因定位

核心问题是异步链路未做完成等待，直接采用了fire-and-forget的订阅模式：原代码在sendMessages内部直接调用.subscribe()，该方法仅完成链路的回调注册，不会阻塞当前执行线程。调用完成后ExecuteQuery的run方法会立刻执行后续逻辑，完全不等KafkaSender完成攒批、发送、等待Broker ACK的全流程，甚至会出现当前线程执行完毕被线程池回收时，后台反应式线程仍在投递消息的情况，最终导致程序提前退出、部分消息未被投递。
成功日志的触发时机存在认知偏差：原代码中的doOnSuccess回调，是在发送流接收到上游源Flux的完成信号时就可能触发，但此时KafkaSender内部的批量缓冲尚未全部发送到Broker，并非真正意义上的“全量投递成功”。
节流逻辑完全无效：代码中写的Thread.sleep(0, 5)为5纳秒级阻塞，既达不到节流效果，还会直接阻塞Reactor的事件循环线程，拖慢整体发送效率。

可靠投递改造方案

结合当前Spring控制台应用、定时调度驱动任务的架构，按以下步骤调整即可保障全量消息可靠送达：

拆分反应式链路的职责，去掉内部隐式订阅
把原来在LogProducer内部subscribe的逻辑移除，直接返回发送流程对应的Mono信号，让调用方自己决定等待策略：

// 改造后的sendMessages方法，返回可阻塞的完成信号，不内部发起订阅
public Mono<Void> sendMessages(String topic, Flux<Logs.Data> records) {
    AtomicInteger sentCount = new AtomicInteger(0);
    return sender.send(records
            // 替换原无效sleep，用Reactor内置背压实现节流
            .limitRate(50)
            .map(record -> {
                LogRecord lrec = record.getRecords().get(0);
                String id = lrec.getId();
                return SenderRecord.create(
                        new ProducerRecord<>(topic, id, lrec.toString()),
                        id
                );
            }))
            .doOnNext(res -> sentCount.incrementAndGet())
            .doOnError(e -> log.error("[FAIL]: 发送到主题'{}'失败", topic, e))
            .doOnSuccess(s -> log.info("[SUCCESS]: 共发送{}条记录到主题'{}'", sentCount.get(), topic))
            .then();
}

调用侧阻塞等待全链路完成，设置合理超时
在ExecuteQuery的执行逻辑里，等发送流程完全拿到Broker ACK后，再执行后续业务逻辑，避免提前退出：

@Override
public void run() {
    Flux<Logs.Data> records = ...;
    try {
        // 阻塞当前线程直到所有消息投递完成，设置10分钟超时避免无限阻塞
        producer.sendMessages(kafkaTopic, records)
                .timeout(Duration.ofMinutes(10))
                .block();
    } catch (Exception e) {
        log.error("消息投递流程异常，终止当前任务后续执行", e);
        return;
    }
    // 只有全量消息成功送达后，才会执行后续逻辑
    // ..... 投递完成后的业务处理代码
}

调整KafkaSender的可靠配置，避免缓冲消息丢失
LogProducer本身应该作为单例复用，不要在每次任务执行时重新初始化，避免重复创建Producer实例造成资源浪费。在初始化SenderOptions时补充配套配置，确保流程退出时内部缓冲的消息能全部flush到Broker：

public LogProducer(String bootstrapServers) {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ProducerConfig.CLIENT_ID_CONFIG, "log-producer");
    props.put(ProducerConfig.ACKS_CONFIG, "all");
    props.put(ProducerConfig.RETRIES_CONFIG, 3); // 发送失败自动重试3次
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    SenderOptions<String, String> senderOptions = SenderOptions.create(props);
    // 设置30秒关闭超时，确保流程结束时内部攒批的消息能全部发完
    senderOptions.closeTimeout(Duration.ofSeconds(30));
    sender = KafkaSender.create(senderOptions);
}

匹配线程池调度的边界
当前架构中定时调度线程仅负责把任务丢给独立线程池执行的逻辑是合理的，需要确保：
- 线程池的核心线程数匹配任务执行的最长耗时，避免因为发送流程阻塞导致线程不够用
- 给定时任务加上并发控制，避免上一轮任务还没完成投递，新的一轮任务就已经启动，造成重复投递

注意：上述block()操作是运行在ExecuteQuery所属的线程池线程上，而非Reactor的核心事件循环线程，不会阻塞反应式框架的调度效率，是Reactor官方推荐的在非反应式代码中接入反应式链路的标准方式。

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