SpringBoot集成kafka-自定义拦截器(可以在拦截器中做记录日志、安全检查等操作)

@TOC

在这里插入图片描述

kafka_6">1、kafka配置类

kafka配置类添加@Configuration注解，springboot启动后会自动读取该配置类；
由于在application.yml文件中我们找不到kafak拦截器相关的配置项，因此需要自定义拦截器；
消费者相关配置方法中添加自定义拦截器配置，这样就可以在自定义拦截器中处理个性化业务需求；
配置类中需要注入消费者工厂bean和消费者监听器工厂，以替换kafak内置默认的消费者工厂和消费者监听器工厂。

package com.power.config;import com.power.Inteceptor.CustomConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;import java.util.HashMap;
import java.util.Map;/*** kafka配置类*/
@Configuration
public class KafkaConfig {@Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers;@Value("${spring.kafka.consumer.key-deserializer}")private String keyDeserializer;@Value("${spring.kafka.consumer.value-deserializer}")private String valueDeserializer;/*** 消费者相关配置* @return*/public Map<String,Object> consumerConfigs(){Map<String,Object> props = new HashMap<>();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,keyDeserializer);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,valueDeserializer);//添加一个消费者拦截器props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, CustomConsumerInterceptor.class.getName());return props;}/*** 消费者工厂*/@Beanpublic ConsumerFactory<String,String> consumerFactory(){return new DefaultKafkaConsumerFactory<>(consumerConfigs());}@Beanpublic KafkaListenerContainerFactory<?> ourKafkaListenerContainerFactory(ConsumerFactory<String,String> ourConsumerFactory){ConcurrentKafkaListenerContainerFactory<String,String> listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory<>();listenerContainerFactory.setConsumerFactory(ourConsumerFactory);return listenerContainerFactory;}}

2、自定义拦截器类

package com.power.Inteceptor;import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;import java.util.Map;/*** 自定义的消费者拦截器*/
public class CustomConsumerInterceptor implements ConsumerInterceptor<String,String> {/*** 在消费消息之前执行* @param record* @return*/@Overridepublic ConsumerRecords onConsume(ConsumerRecords record) {System.out.println("onConsumer方法执行（在消费消息之前执行），record="+record);return record;}/*** 在拿到消息之后，提交offset之前执行该方法* @param offsets*/@Overridepublic void onCommit(Map offsets) {System.out.println("onCommit方法执行（在拿到消息之后，提交offset之前执行该方法），offsets="+offsets);}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}

3、消费者

package com.power.consumer;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;@Component
public class EventConsumer {@KafkaListener(topics = {"interceptorTopic"}, groupId = "interceptorGroup", containerFactory = "ourKafkaListenerContainerFactory")public void onEvent(ConsumerRecord<String, String> record) {System.out.println("消费者消费消息record = " + record);}
}

4、生产者

package com.power.producer;import com.power.model.User;
import com.power.util.JSONUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.Date;@Component
public class EventProducer {@Resourceprivate KafkaTemplate<String,Object> kafkaTemplate;public void sendEvent(){User user = User.builder().id(1).phone("15676767673").birthday(new Date()).build();String userJson = JSONUtils.toJSON(user);kafkaTemplate.send("interceptorTopic","k", userJson);}}

5、实体类（用于发送接收对象消息）

package com.power.model;import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;import java.util.Date;@Builder
@AllArgsConstructor
@NoArgsConstructor
@Data
public class User {private Integer id;private String phone;private Date birthday;}

6、JSON工具类

package com.power.util;import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;public class JSONUtils {private static final ObjectMapper OBJECTMAPPER = new ObjectMapper();public static String toJSON(Object object){try {return OBJECTMAPPER.writeValueAsString(object);} catch (JsonProcessingException e) {throw new RuntimeException(e);}}public static <T> T toBean(String json,Class<T> clazz){try {return OBJECTMAPPER.readValue(json,clazz);} catch (JsonProcessingException e) {throw new RuntimeException(e);}}
}

7、启动类

package com.power;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;import java.util.Map;@SpringBootApplication
public class Kafka04Application {public static void main(String[] args) {ConfigurableApplicationContext context = SpringApplication.run(Kafka04Application.class, args);Map<String, ConsumerFactory> beansOfType = context.getBeansOfType(ConsumerFactory.class);beansOfType.forEach((k,v)->{System.out.println(k+" -- "+v);});System.out.println("----------------------------------------------------");Map<String, KafkaListenerContainerFactory> beansOfType2 = context.getBeansOfType(KafkaListenerContainerFactory.class);beansOfType2.forEach((k,v)->{System.out.println(k+" -- "+v);});}}

以下红框内容用于查看SpringBoot启动后注入的类型
在这里插入图片描述

8、测试类

package com.power;import com.power.producer.EventProducer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;import javax.annotation.Resource;@SpringBootTest
public class SpringBoot04KafkaBaseApplication {@Resourceprivate EventProducer eventProducer;@Testvoid sendInterceptor(){eventProducer.sendEvent();}}