Reactor Kafka优雅停机与消费者暂停及WebClient异常咨询

你当前用@PreDestroy加硬编码Thread.sleep的方式并不靠谱——sleep时长完全靠拍脑袋，无法适配不同消息的处理耗时。Reactor Kafka结合Spring生命周期的优雅停机标准做法是管理消费流的生命周期，并跟踪正在处理的任务，确保所有在途任务完成后再销毁，具体步骤如下：

1. 保存消费流的订阅引用：启动消费时，将流的Disposable或Mono保存下来，用于后续控制生命周期。
2. 暂停消费+停止接收新消息：先暂停Kafka消费者的分区分配，避免拉取新消息；再取消消费流的订阅，停止接收新的消息事件。
3. 等待在途任务完成：通过跟踪处理中的任务（比如用计数器）或利用Reactor的流特性，等待所有正在执行的消息处理逻辑完成，而非硬编码sleep。

改进后的代码示例：

private Disposable consumerDisposable;
private final AtomicInteger processingTaskCount = new AtomicInteger(0);

@PostConstruct
public void startConsumer() {
    consumerDisposable = reactiveKafkaConsumerTemplate
            .receiveAutoAck()
            .publishOn(Schedulers.boundedElastic())
            .flatMap(x -> Mono.just(x)
                    .delayElement(Duration.ofMillis(300)), 5)
            .flatMap(message -> {
                processingTaskCount.incrementAndGet();
                return processMessageImp.processMessage(message)
                        .doFinally(signalType -> processingTaskCount.decrementAndGet())
                        .onErrorResume(t -> {
                            log.error("消息处理失败", t);
                            processingTaskCount.decrementAndGet();
                            return Mono.empty();
                        });
            })
            .subscribe();
}

public void stopConsumerGracefully() {
    // 1. 暂停Kafka消费者，不再拉取新消息
    reactiveKafkaConsumerTemplate.assignment()
            .flatMap(reactiveKafkaConsumerTemplate::pause)
            .block();

    // 2. 取消订阅，停止接收新的消息事件
    consumerDisposable.dispose();

    // 3. 等待所有在途任务处理完成
    while (processingTaskCount.get() > 0) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
        }
    }
}

@PreDestroy
public void onExit() {
    stopConsumerGracefully();
}

这个异常的根源是：Spring注入的WebClient.Builder默认使用Spring管理的Reactor Netty线程池/连接池，在上下文销毁阶段会提前关闭这些资源。你用sleep的方式无法保证所有WebClient请求在资源关闭前完成，导致后续请求抛出"executor not accepting a task"异常。

解决思路有两种：

方案1：确保所有WebClient请求在资源销毁前完成 #

基于问题1的优雅停机逻辑，先等待所有消息处理（包括WebClient请求）完成，再让Spring销毁WebClient相关资源。

修改消费流为Mono<Void>，利用Reactor的then()方法等待所有处理完成：

private Mono<Void> consumerMono;

@PostConstruct
public void startConsumer() {
    consumerMono = reactiveKafkaConsumerTemplate
            .receiveAutoAck()
            .publishOn(Schedulers.boundedElastic())
            .flatMap(x -> Mono.just(x)
                    .delayElement(Duration.ofMillis(300)), 5)
            .flatMap(message -> processMessageImp.processMessage(message)
                    .onErrorResume(t -> {
                        log.error("消息处理失败", t);
                        return Mono.empty();
                    }))
            .then(); // 流结束标记：所有消息处理完成后才会完成
    consumerMono.subscribe();
}

@PreDestroy
public void onExit() {
    // 暂停Kafka消费
    reactiveKafkaConsumerTemplate.assignment()
            .flatMap(reactiveKafkaConsumerTemplate::pause)
            .block();

    // 等待所有消息处理（包括WebClient请求）完成
    consumerMono.block();
}

方案2：自定义WebClient的资源池，手动控制销毁时机 #

如果不想依赖Spring的资源管理，可以自己创建WebClient的线程池，在所有请求完成后再手动关闭：

private NioEventLoopGroup webClientEventLoopGroup;

public WebClient createWebclient() {
    webClientEventLoopGroup = new NioEventLoopGroup();
    HttpClient httpClient = HttpClient.create()
            .tcpConfiguration(tcpClient -> tcpClient
                    .bootstrap(server -> server
                            .group(webClientEventLoopGroup)
                            .channel(NioSocketChannel.class)));

    return builder
            .clientConnector(new ReactorClientHttpConnector(httpClient))
            .build();
}

@PreDestroy
public void onExit() {
    stopConsumerGracefully(); // 先等待所有消息处理完成
    // 优雅关闭WebClient的线程池
    webClientEventLoopGroup.shutdownGracefully().syncUninterruptibly();
}

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