SpringBoot-Kafka使用（五）

水深无声 2022-03-07 07:44 259阅读 0赞

## 一、消息过滤器 ##

消息过滤器

消息过滤器可以在消息抵达监听容器前被拦截，过滤器根据系统业务逻辑去筛选出需要的数据再交由KafkaListener处理。

配置消息其实是非常简单的额，只需要为监听容器工厂配置一个RecordFilterStrategy(消息过滤策略)，返回true的时候消息将会被抛弃，返回false时，消息能正常抵达监听容器。

这里我们将消息转换为long类型，判断该消息为基数还是偶数，把所有基数过滤，监听容器只接收偶数。

@Component
    public class FilterListener {
    
        private static final Logger log= LoggerFactory.getLogger(TaskListener.class);
    
        @Autowired
        private ConsumerFactory consumerFactory;
    
        @Bean
        public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
            ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
            factory.setConsumerFactory(consumerFactory);
            //配合RecordFilterStrategy使用，被过滤的信息将被丢弃
            factory.setAckDiscarded(true);
            factory.setRecordFilterStrategy(new RecordFilterStrategy() {
                @Override
                public boolean filter(ConsumerRecord consumerRecord) {
                    long data = Long.parseLong((String) consumerRecord.value());
                    log.info("filterContainerFactory filter : "+data);
                    if (data % 2 == 0) {
                        return false;
                    }
                    //返回true将会被丢弃
                    return true;
                }
            });
            return factory;
        }
    
        @KafkaListener(id = "filterCons", topics = "topic.quick.filter",containerFactory = "filterContainerFactory")
        public void filterListener(String data) {
            //这里做数据持久化的操作
            log.error("topic.quick.filter receive : " + data);
        }
    }

测试方法中，我们将当前时间戳发送到Kafka中。

@Test
        public void testFilter() throws InterruptedException {
            kafkaTemplate.send("topic.quick.filter", System.currentTimeMillis()+"");
        }

## 二、异常处理器 ##

### 异常处理 ###

平常程序异常我们使用try catch捕获异常，在catch方法中根据异常类型进行相关处理，既然我们可以使用try catch处理异常，那为什么还要使用ConsumerAwareErrorHandler异常处理器去处理异常呢？  
首先，KafkaListener要做的事只是监听Topic中的数据并消费，如果在KafkaListener中还需要对异常进行处理则会显得代码块非常臃肿不利于维护，我们可以把异常处理的这些代码抽象出来，构造成一个异常处理器，KafkaListener中所抛出的异常都会经过ConsumerAwareErrorHandler异常处理器进行处理，这样就非常方便我们进行后期维护，比如后期更改异常处理业务的时候，只需要修改ConsumerAwareErrorHandler处理器就行了，而不需要KafkaListener的一堆代码中去修改代码。这也是一种思想的体现。

### 单消息消费异常处理器 ###

这里主要就是注册一个ConsumerAwareListenerErrorHandler 类型的异常处理器，bean的注册默认使用的是方法名，所以我们将这个异常处理的BeanName放到@KafkaListener注解的errorHandler属性里面。当KafkaListener抛出异常的时候，则会自动调用异常处理器。

@Component
    public class ErrorListener {
    
        private static final Logger log= LoggerFactory.getLogger(ErrorListener.class);
    
        @KafkaListener(id = "err", topics = "topic.quick.error", errorHandler = "consumerAwareErrorHandler")
        public void errorListener(String data) {
            log.info("topic.quick.error  receive : " + data);
            throw new RuntimeException("fail");
        }
    
        @Bean
        public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
            return new ConsumerAwareListenerErrorHandler() {
    
                @Override
                public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {
                    log.info("consumerAwareErrorHandler receive : "+message.getPayload().toString());
                    return null;
                }
            };
        }
    
    }

编写测试方法，发送一条消息到topic.quick.error中，运行测试方法后我们可以看到异常处理器已经能正常使用了。

@Autowired
        private KafkaTemplate kafkaTemplate;
    
        @Test
        public void testErrorHandler() {
            kafkaTemplate.send("topic.quick.error", "test error handle");
        }

2018-09-14 11:42:05.099  INFO 8912 --- [      err-0-C-1] com.viu.kafka.listen.ErrorListener       : topic.quick.error  receive : test error handle
    2018-09-14 11:42:05.101  INFO 8912 --- [      err-0-C-1] com.viu.kafka.listen.ErrorListener       : consumerAwareErrorHandler receive : test error handle

### 批量消费异常处理器 ###

批量消费代码也是差不多的，只不过传递过来的数据都是List集合方式，这里就不做其他代码的展示了。

@Bean
        public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
            return new ConsumerAwareListenerErrorHandler() {
    
                @Override
                public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {
                    log.info("consumerAwareErrorHandler receive : "+message.getPayload().toString());
                    MessageHeaders headers = message.getHeaders();
                    List<String> topics = headers.get(KafkaHeaders.RECEIVED_TOPIC, List.class);
                    List<Integer> partitions = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, List.class);
                    List<Long> offsets = headers.get(KafkaHeaders.OFFSET, List.class);
                    Map<TopicPartition, Long> offsetsToReset = new HashMap<>();
              
                    return null;
                }
            };
        }