SpringBoot整合kafka

Love The Way You Lie 2023-01-17 14:56 218阅读 0赞

一：SpringBoot整合kafka

引入kafka依赖

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

application.properties配置

###########【Kafka集群】###########
    spring.kafka.bootstrap-servers=192.168.10.136:9092
    ###########【初始化生产者配置】###########
    # 重试次数
    spring.kafka.producer.retries=0
    # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
    spring.kafka.producer.acks=1
    # 批量大小
    spring.kafka.producer.batch-size=16384
    # 提交延时
    spring.kafka.producer.properties.linger.ms=0
    # 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
    # linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了
    
    # 生产端缓冲区大小
    spring.kafka.producer.buffer-memory = 33554432
    # Kafka提供的序列化和反序列化类
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
    # 自定义分区器
    # spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner
    
    ###########【初始化消费者配置】###########
    # 默认的消费组ID
    spring.kafka.consumer.properties.group.id=defaultConsumerGroup
    # 是否自动提交offset
    spring.kafka.consumer.enable-auto-commit=true
    # 提交offset延时(接收到消息后多久提交offset)
    spring.kafka.consumer.auto.commit.interval.ms=1000
    # 当kafka中没有初始offset或offset超出范围时将自动重置offset
    # earliest:重置为分区中最小的offset;
    # latest:重置为分区中最新的offset(消费分区中新产生的数据);
    # none:只要有一个分区不存在已提交的offset,就抛出异常;
    spring.kafka.consumer.auto-offset-reset=latest
    # 消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
    spring.kafka.consumer.properties.session.timeout.ms=120000
    # 消费请求超时时间
    spring.kafka.consumer.properties.request.timeout.ms=180000
    # Kafka提供的序列化和反序列化类
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    # 消费端监听的topic不存在时，项目启动会报错(关掉)
    spring.kafka.listener.missing-topics-fatal=false
    # 设置批量消费
    # spring.kafka.listener.type=batch
    # 批量消费每次最多消费多少条消息
    # spring.kafka.consumer.max-poll-records=50

KafkaProducer生产者：

package com.example.kafka_demo.producer;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RestController;
    
    /**
     * @author lizhangyu
     * @date 2021/5/1 16:10
     */
    @RestController
    public class KafkaProducer {
    
        /**
         * 发送消息
         */
        @Autowired
        private KafkaTemplate<String, Object> kafkaTemplate;
    
        /**
         * 发送消息
         * @param normalMessage
         */
        @GetMapping("/kafka/normal/{message}")
        public void sendMessage1(@PathVariable("message") String normalMessage) {
            kafkaTemplate.send("topic", normalMessage);
        }
    
    }

KafkaConsumer消费者：

package com.example.kafka_demo.consumer;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    /**
     * @author lizhangyu
     * @date 2021/5/1 16:28
     */
    @Component
    public class KafkaConsumer {
    
        /**
         * 消费监听
         * @param record
         */
        @KafkaListener(topics = {"topic"})
        public void onMessage(ConsumerRecord<?, ?> record){
            // 消费的哪个topic、partition的消息,打印出消息内容
            System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
        }
    
    }

使用postman发送请求：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NDY5MDU1_size_16_color_FFFFFF_t_70][]

运行结果：

2021-05-01 23:07:47.724  INFO 15972 --- [nio-8080-exec-2] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 2.6.0
    2021-05-01 23:07:47.724  INFO 15972 --- [nio-8080-exec-2] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 62abe01bee039651
    2021-05-01 23:07:47.724  INFO 15972 --- [nio-8080-exec-2] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1619881667724
    2021-05-01 23:07:47.735  INFO 15972 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Cluster ID: VJuFZZpiRGqdhW58t8Bn3g
    简单消费：topic-0-HelloKafka

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NDY5MDU1_size_16_color_FFFFFF_t_70]: /images/20221021/b2e2a967a64a4635bd251fb240a4f123.png