@TOC
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;
}
/**
* 消费者工厂
*/
@Bean
public ConsumerFactory<String,String> consumerFactory(){
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public 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
*/
@Override
public ConsumerRecords onConsume(ConsumerRecords record) {
System.out.println("onConsumer方法执行（在消费消息之前执行），record="+record);
return record;
}
/**
* 在拿到消息之后，提交offset之前执行该方法
* @param offsets
*/
@Override
public void onCommit(Map offsets) {
System.out.println("onCommit方法执行（在拿到消息之后，提交offset之前执行该方法），offsets="+offsets);
}
@Override
public void close() {
}
@Override
public 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 {
@Resource
private 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 {
@Resource
private EventProducer eventProducer;
@Test
void sendInterceptor(){
eventProducer.sendEvent();
}
}
9、测试
- 先启动消费者监听
- 在启动生产者发送消息
- 测试结果发现，消费者走了我们自定义的拦截器
