【弄nèng-Kafka】应用篇(八)——Springboot整合Kafka(异常处理器

2024年2月6日发(作者：)

import red;import ier;import ;import emplate;import ent;@Componentpublic class ErrorProducer { @Autowired @Qualifier("errorTemplate") private KafkaTemplate kafkaTemplate; @Value("${}") private String topic; public void send(String message) { (topic, message); }}2.2.2 消费者配置类

import erConfig;import Deserializer;import ;import ;import uration;import Kafka;import rentKafkaListenerContainerFactory;import istenerContainerFactory;import tKafkaConsumerFactory;import p;import ;@Configuration@EnableKafkapublic class ErrorConsumerConfig { @Value("${rap-servers}") private String bootstrapServers; @Value("${}") private String topic; /** *

单线程-单条消费 * @return */ @Bean public KafkaListenerContainerFactory<?> errorKafkaListenerContainerFactory() { Map configProps = new HashMap<>(); (RAP_SERVERS_CONFIG, bootstrapServers); (_DESERIALIZER_CLASS_CONFIG, ); (_DESERIALIZER_CLASS_CONFIG, ); (_ID_CONFIG, topic); ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); sumerFactory(new DefaultKafkaConsumerFactory<>(configProps)); return factory; }}消费者

import 4j;import er;import ;import istener;import erAwareListenerErrorHandler;import erExecutionFailedException;import eaders;import e;import eHeaders;import ;import d;import ent;import ;@Component@Slf4jpublic class ErrorConsumer { @KafkaListener(topics = "${}", containerFactory = "errorKafkaListenerContainerFactory", errorHandler = "consumerAwareErrorHandler") public void receive(@Payload String message, @Header(ED_PARTITION_ID) int partition) { n(("From partition %d : %s", partition, message) ); throw new RuntimeException("fail"); } /** *

单条消息 * @return */ @Bean public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() { return new ConsumerAwareListenerErrorHandler() { @Override public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) { ("ConsumerAwareListenerErrorHandler receive : "+load().toString()); return null; } }; } /** *

批量消息 * @return */ @Bean public ConsumerAwareListenerErrorHandler consumerAwareErrorHandlerBatch() { return new ConsumerAwareListenerErrorHandler() { @Override public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) { ("consumerAwareErrorHandler receive : "+load().toString()); MessageHeaders headers = ders(); List topics = (ED_TOPIC, ); List partitions = (ED_PARTITION_ID, ); List offsets = (, ); return null; } }; }}