Spring Boot整合Kafka

2021-11-18 05:06:31

Spring Boot整合Kafka

pom.xml

<dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

默认使用

配置

prop新增kafka配置详细配置可参考org.springframework.boot.autoconfigure.kafka.KafkaProperties中属性生产者,消费者默认序列化都是String格式message

#kafka默认消费者配置
spring.kafka.consumer.bootstrap-servers=10.111.7.124:9092
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-offset-reset=earliest
#kafka默认生产者配置
spring.kafka.producer.bootstrap-servers=10.111.7.124:9092
spring.kafka.producer.acks=-1
spring.kafka.client-id=kafka-producer
spring.kafka.producer.batch-size=5

使用

//生产者
@Resource
private KafkaTemplate kafkaTemplate;
public void send() {
    HashMap<String, String> map = new HashMap<>();
    map.put("sendType","send");
    kafkaTemplate.send("test01", JSONUtil.toJsonStr(map));
}

//消费者
@Slf4j
@Component
public class KafkaConsumer {
    @KafkaListener(topics = "test01",groupId = "group01")
    public void listen(String message) {
        log.info("message:{}",message);
    }
}

使用Json格式化

配置

自定义消费者工厂: 配置全部消费者参数,包含转换器,使用此项可删除转换器配置2

**转换器配置2:**仅配置json转换器,不需要配置全部消费者参数时可删除自定义消费者配置

addTrustedPackages("*"):默认Json转换仅信任java.util,java.lang包下类,增加后不进行类型检查

@Configuration
public class KafkaCustomConfig {
    @Value("${spring.kafka.consumer.bootstrap-servers}")
    private String bootstrapServers;
    //自定义消费者工厂
    @Bean("customContainerFactory")
    public ConcurrentKafkaListenerContainerFactory customContainerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "customGroup01");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        //转换器配置1
        DefaultKafkaConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory(props);
        JsonDeserializer jsonDeserializer = new JsonDeserializer();
        jsonDeserializer.getTypeMapper().addTrustedPackages("*");
        consumerFactory.setValueDeserializer(jsonDeserializer);
        //指定使用DefaultKafkaConsumerFactory
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        //设置可批量拉取消息消费，拉取数量一次3，看需求设置
        factory.setConcurrency(3);
        factory.setBatchListener(true);
        return factory;
    }
    //转换器配置2
    @Bean
    public RecordMessageConverter converter() {
        ByteArrayJsonMessageConverter converter = new ByteArrayJsonMessageConverter();
        DefaultJackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
        typeMapper.setTypePrecedence(Jackson2JavaTypeMapper.TypePrecedence.TYPE_ID);
        typeMapper.addTrustedPackages("*");
        converter.setTypeMapper(typeMapper);
        return converter;
    }
    /**
     * 不使用spring boot的KafkaAutoConfiguration默认方式创建的KafkaTemplate，重新定义
     * 与默认配置只能存在一个
     * @return
     */
    @Bean("custiomKafkaTemplate")
    public KafkaTemplate custiomKafkaTemplate() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        //0 producer不等待broker同步完成的确认，继续发送下一条(批)信息
        //1 producer要等待leader成功收到数据并得到确认，才发送下一条message。
        //-1 producer得到follwer确认，才发送下一条数据
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 5);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 500);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        DefaultKafkaProducerFactory produceFactory = new DefaultKafkaProducerFactory(props);
        return new KafkaTemplate(produceFactory);
    }
}

使用

@Data
public class TestVo {
    private String key;
    private String value;
}

生产者

//注入自定义KafkaTemplate
@Resource(name = "custiomKafkaTemplate")
private KafkaTemplate custiomKafkaTemplate;
public void customSend() {
    TestVo vo = new TestVo();
    vo.setKey("sendType");
    vo.setValue("send");
    custiomKafkaTemplate.send("custom04", vo);
}

消费者

containerFactory:指定自定义消费者工厂beanName

@Slf4j
@Component
public class KafkaConsumer {
    //自定义消费者
    @KafkaListener(topics = "custom04",containerFactory = "customContainerFactory")
    public void customListen(TestVo message) {
        log.info("message:{}",message.toString());
    }
    //仅自定义转换器
    @KafkaListener(topics = "custom04",groupId = "custom04Group")
    public void customListen(TestVo message) {
        log.info("message:{}",message.toString());
    }
}

参考资料

https://github.com/spring-projects/spring-kafka/tree/master/samples

https://blog.csdn.net/u012045045/article/details/111034500

https://www.tutorialspoint.com/spring_boot/spring_boot_apache_kafka.htm

码农公寓

Spring Boot整合Kafka

默认使用

配置

使用

使用Json格式化

配置

使用

参考资料

相关文章