spring boot 全局异常处理无法直接捕获 kafka 消费者内部的重平衡（rebalance）异常，因为重平衡是发生在客户端线程中的异步事件，而非同步抛出的 java 异常。要解决这个问题，必须实现 kafkalistenererrorhandler 接口或注册 consumerawarelistenererrorhandler，在监听器方法中通过返回错误对象或使用 throw new listenerexecutionfailedexception 来触发全局异常处理器。
如何配置监听器以支持全局异常捕获
要实现全局捕获，需要在 @KafkaListener 注解中指定 errorHandler 属性，指向一个实现了 KafkaListenerErrorHandler 的 Bean。当消费者在处理消息过程中发生异常，或者在重平衡期间出现错误时，该处理器会被调用。如果处理器抛出异常，Spring 会将其传递给全局异常处理器（如 @ControllerAdvice 或 @RestControllerAdvice）。关键在于，重平衡本身不会直接抛出异常到业务逻辑层，但如果在重平衡后重新分配分区并拉取消息时发生错误，这些错误可以通过上述机制捕获。
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.stereotype.Component;
@Component
public class MyKafkaListener {
@KafkaListener(topics = "my-topic", errorHandler = "kafkaErrorHandler")
public void listen(ConsumerRecord<String, String> record) {
// 模拟处理逻辑，可能在重平衡后执行时出错
if (record.value().equals("error")) {
throw new RuntimeException("Processing failed");
}
System.out.println("Processed: " + record.value());
}
}
上述代码定义了监听器，并指定了 kafkaErrorHandler。接下来需要定义这个错误处理器。
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.support.KafkaUtils;
import org.springframework.stereotype.Component;
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
@Component("kafkaErrorHandler")
public class CustomKafkaErrorHandler implements KafkaListenerErrorHandler {
@Override
public Object handleError(Message<?> message, ListenerExecutionFailedException exception) {
// 这里可以记录日志或进行其他处理
// 关键：抛出异常以便被全局异常处理器捕获
throw new ListenerExecutionFailedException(
"Kafka listener error",
exception,
KafkaUtils.buildMessageInfo((ConsumerRecord<?, ?>) message.getPayload())
);
}
}
最后，在全局异常处理器中捕获 ListenerExecutionFailedException。
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;
import org.springframework.http.ResponseEntity;
import org.springframework.http.HttpStatus;
@RestControllerAdvice
public class GlobalExceptionHandler {
@ExceptionHandler(ListenerExecutionFailedException.class)
public ResponseEntity<String> handleKafkaError(ListenerExecutionFailedException ex) {
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body("Kafka processing failed: " + ex.getMessage());
}
}
运行此程序时，如果发送值为 "error" 的消息，将触发处理异常，并通过全局处理器返回 HTTP 500 响应。重平衡期间的错误通常会在消费者重启或重新分配分区时发生，此时如果处理逻辑出错，同样会被捕获。
重平衡监听器与全局异常的交互
除了直接捕获处理异常，还可以使用 ConsumerAwareListenerErrorHandler 来获取消费者实例，从而在重平衡前后执行自定义逻辑。虽然重平衡本身不抛出异常，但可以在重平衡回调中检查状态或记录信息。如果需要将重平衡事件视为一种“异常”情况并进行全局处理，可以在回调中抛出异常，但这通常不是推荐做法，因为重平衡是正常行为。更常见的做法是结合 ConsumerRebalanceListener 来监控重平衡，并在必要时调整业务逻辑。
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
@Component("consumerAwareErrorHandler")
public class ConsumerAwareErrorHandler implements ConsumerAwareListenerErrorHandler {
@Override
public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
// 可以访问消费者实例，例如提交偏移量或记录重平衡信息
System.out.println("Consumer rebalance or error occurred. Partitions: " + consumer.assignment());
throw new ListenerExecutionFailedException("Consumer aware error", exception);
}
}
在监听器中使用 errorHandler = "consumerAwareErrorHandler"。这样，当发生错误或重平衡相关的问题时，可以获取消费者实例，从而更好地管理状态。注意，重平衡事件本身不会导致 handleError 被调用，除非在处理消息时发生异常。如果需要专门处理重平衡事件，应使用 ConsumerRebalanceListener，而不是依赖异常处理机制。
如何处理重平衡导致的重复消费问题
重平衡可能导致消息被重复消费，因为偏移量可能在重平衡前未提交。全局异常处理无法直接解决重复消费问题，但可以在异常处理中记录重试次数或幂等性键。建议在业务逻辑中实现幂等性，例如使用数据库唯一约束或 Redis 原子操作。如果必须在异常处理中处理重复消费，可以在 GlobalExceptionHandler 中检查消息的唯一标识，避免重复处理。
