动态停止单台服务器上的Spring Kafka Consumer

1. 用KafkaListenerEndpointRegistry动态启停Consumer #

直接在应用中注入KafkaListenerEndpointRegistry，它可以管理所有Kafka监听容器。编写简单接口触发单台服务器上所有Consumer的暂停/停止操作：

@RestController
public class ConsumerControlController {
    @Autowired
    private KafkaListenerEndpointRegistry kafkaListenerRegistry;

    // 暂停当前服务器所有Consumer（停止拉取消息，但保持与Broker的连接）
    @PostMapping("/pause-kafka-consumers")
    public void pauseAllConsumers() {
        kafkaListenerRegistry.getListenerContainers().forEach(container -> {
            if (container.isRunning()) {
                container.pause();
            }
        });
    }

    // 彻底停止当前服务器所有Consumer（断开连接，触发Broker重平衡）
    @PostMapping("/stop-kafka-consumers")
    public void stopAllConsumers() {
        kafkaListenerRegistry.getListenerContainers().forEach(container -> {
            if (container.isRunning()) {
                container.stop();
            }
        });
    }

    // 恢复Consumer运行
    @PostMapping("/resume-kafka-consumers")
    public void resumeAllConsumers() {
        kafkaListenerRegistry.getListenerContainers().forEach(container -> {
            if (container.isPaused()) {
                container.resume();
            } else if (!container.isRunning()) {
                container.start();
            }
        });
    }
}

调用目标服务器的/stop-kafka-consumers接口，就能让该机器上的3个Consumer彻底停止，仅保留另一台服务器的3个Consumer运行。如果只是临时暂停不想断开连接，调用/pause-kafka-consumers即可。

2. 用配置开关控制Consumer是否初始化 #

在配置文件中添加自定义开关，比如app.kafka.consumer.enabled=true，然后给Kafka监听方法添加条件注解：

@KafkaListener(topics = "your-target-topic", concurrency = "3")
@ConditionalOnProperty(name = "app.kafka.consumer.enabled", havingValue = "true", matchIfMissing = true)
public void processMessage(ConsumerRecord<String, Object> record) {
    // 消息处理逻辑
}

需要停止某台服务器的Consumer时，将该机器的配置改为app.kafka.consumer.enabled=false：

• 若使用配置中心（如Nacos、Spring Cloud Config），直接动态刷新配置即可，无需重启容器；
• 未使用配置中心的话，修改配置后重启应用，此时该机器不会初始化那3个Consumer，仅另一台服务器的Consumer正常运行。

3. 用事件机制触发Consumer暂停 #

如果需要内部逻辑自动触发暂停，可以发布ConsumerPauseEvent事件：

@Autowired
private ApplicationEventPublisher eventPublisher;
@Autowired
private KafkaListenerEndpointRegistry kafkaListenerRegistry;

public void pauseConsumersByEvent() {
    kafkaListenerRegistry.getListenerContainers().forEach(container -> {
        eventPublisher.publishEvent(new ConsumerPauseEvent(this, container.getGroupId(), container.getAssignedPartitions()));
    });
}

效果与方案1的暂停功能一致，适合无需外部接口触发的场景。

注意事项 #

• 暂停Consumer不会触发Broker重平衡，Consumer仍属于消费组；只有调用stop()彻底停止，Broker才会触发重平衡，将分区全部分配给其他运行的Consumer。
• 若使用动态刷新配置，需确保应用开启了Actuator的refresh端点，否则需要重启应用才能让配置生效。

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