KAFKA-05-kafkaspringboot集成

朴灿烈づ我的快乐病毒、 2023-09-29 14:06 41阅读 0赞

### 文章目录 ###

*  1：引入依赖
 *  2：配置application.properties
 *   *  1:常规配置
     *  2:全部配置含义
 *  3：hello kakfa-简单应用
 *   *  1：简单生产者
     *  2：简单消费者
     *  3:测试结果
 *  4：生产者
 *   *  1：带回调的生产者
     *  2：事务提交
 *  5：消费者
 *   *  1、指定topic、partition、offset消费
     *  2、批量消费
     *  3:ConsumerAwareListenerErrorHandler 异常处理器
     *  4:消费者&生产者--消息转发
     *  5:消息拦截器
     *  6、定时启动、停止监听器

# 1：引入依赖 #

<parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.1.5.RELEASE</version>
    </parent>
    <dependencies>
    
          <dependency>
               <groupId>org.springframework.boot</groupId>
               <artifactId>spring-boot-starter</artifactId>
           </dependency>
    
           <dependency>
               <groupId>org.springframework.kafka</groupId>
               <artifactId>spring-kafka</artifactId>
           </dependency>
           
           <dependency>
               <groupId>org.springframework.boot</groupId>
               <artifactId>spring-boot-starter-test</artifactId>
               <scope>test</scope>
           </dependency>
    
    </dependencies>

# 2：配置application.properties #

## 1:常规配置 ##

spring.kafka.bootstrap-servers=CentOSA:9092,CentOSB:9092,CentOSC:9092
    
    spring.kafka.producer.retries=5
    spring.kafka.producer.acks=all
    spring.kafka.producer.batch-size=16384
    spring.kafka.producer.buffer-memory=33554432
    spring.kafka.producer.transaction-id-prefix=transaction-id-
    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.enable.idempotence=true
    
    
    spring.kafka.consumer.group-id=group1
    spring.kafka.consumer.auto-offset-reset=earliest
    spring.kafka.consumer.enable-auto-commit=true
    spring.kafka.consumer.auto-commit-interval=100
    spring.kafka.consumer.properties.isolation.level=read_committed
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

## 2:全部配置含义 ##

###########【Kafka集群】###########
    spring.kafka.bootstrap-servers=112.126.74.249:9092,112.126.74.249:9093
    ###########【初始化生产者配置】###########
    # 重试次数
    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

# 3：hello kakfa-简单应用 #

## 1：简单生产者 ##

package com.example.kafkaspringboot.kafkastart;
    
    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;
    
    /**
     * Description:简单的生产者
     * Date:       2022/5/11 - 上午 12:30
     * author:     wangkanglu
     * version:    V1.0
     */
    @RestController
    public class KafkaProduser {
        
    
        @Autowired
        private KafkaTemplate<String,String> kafkaTemplate;
    
        @GetMapping("/kafka/normal/{message}")
        public void sendMessage1(@PathVariable("message") String normalMessage) {
        
            kafkaTemplate.send("topic01", normalMessage);
        }
    
    
    }

## 2：简单消费者 ##

package com.example.kafkaspringboot.kafkastart;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    /**
     * Description:
     * Date:       2022/5/11 - 上午 12:34
     * author:     wangkanglu
     * version:    V1.0
     */
    @Component
    public class KafkaCousumer {
        
        @KafkaListener(topics = {
        "topic01"})
        public void onMessage1(ConsumerRecord<?, ?> record){
        
            // 消费的哪个topic、partition的消息,打印出消息内容
            System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
        }
    }

## 3:测试结果 ##

当启动服务后，消费者就已经处在等待消息队列中  
![在这里插入图片描述][dbecd0f53bf64139bb6ee0b26db4590f.png]  
当请求服务，向topic中传入消息后  
![在这里插入图片描述][e2d155672c9e4cbf8b236df08f8edc1f.png]

# 4：生产者 #

## 1：带回调的生产者 ##

kafkaTemplate提供了一个回调方法addCallback，我们可以在回调方法中监控消息是否发送成功 或 失败时做补偿处理，有两种写法

@Autowired
        private KafkaTemplate<String, Object> kafkaTemplate;
    
        @GetMapping("/kafka/calllback1/{message}")
        public void callBackone(@PathVariable("message") String callbakemessage) {
        
            kafkaTemplate.send("topic01", callbakemessage).addCallback(success -> {
        
                // 消息发送到的topic
                String topic = success.getRecordMetadata().topic();
                // 消息发送到的分区
                int partition = success.getRecordMetadata().partition();
                // 消息在分区内的offset
                long offset = success.getRecordMetadata().offset();
                System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
            }, failure -> {
        
                System.out.println("发送消息失败:" + failure.getMessage());
            });
        }

@Autowired
        private KafkaTemplate<String, Object> kafkaTemplate;
    
     @GetMapping("/kafka/calllback2/{message}")
        public void callBacktwo(@PathVariable("message") String callbakemessage) {
        
            kafkaTemplate.send("topic01", callbakemessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
        
                @Override
                public void onFailure(Throwable ex) {
        
                    System.out.println("发送消息失败：" + ex.getMessage());
                }
    
                @Override
                public void onSuccess(SendResult<String, Object> result) {
        
                    System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
                            + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
                }
            });
        }

## 2：事务提交 ##

如果在发送消息时需要创建事务，可以使用 KafkaTemplate 的 executeInTransaction 方法来声明事务，

package com.example.kafkaspringboot.kafkaProducer;
    
    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;
    
    /**
     * Description:
     * Date:       2022/5/11 - 上午 1:07
     * author:     wangkanglu
     * version:    V1.0
     */
    @RestController
    public class PerduserTransaction {
        
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        @GetMapping("/kafka/transaction/{message}")
        public void transaction(@PathVariable String message) {
        
            // 声明事务：后面报错消息不会发出去
            kafkaTemplate.executeInTransaction(operations -> {
        
                operations.send("topic1", message);
                throw new RuntimeException("fail");
            });
    
        }
    
        @GetMapping("/kafka/untransaction/{message}")
        public void untransaction(@PathVariable String message) {
        
            // 声明事务：后面报错消息不会发出去
            kafkaTemplate.executeInTransaction(operations -> {
        
                operations.send("topic1", message);
                throw new RuntimeException("fail");
            });
    
        }
    }

# 5：消费者 #

## 1、指定topic、partition、offset消费 ##

前面我们在监听消费topic1的时候，监听的是topic1上所有的消息，如果我们想指定topic、指定partition、指定offset来消费呢？也很简单，@KafkaListener注解已全部为我们提供，

package com.example.kafkaspringboot.kafkaConsumer;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.kafka.annotation.PartitionOffset;
    import org.springframework.kafka.annotation.TopicPartition;
    import org.springframework.stereotype.Component;
    
    /**
     * Description:
     * Date:       2022/5/11 - 上午 1:14
     * author:     wangkanglu
     * version:    V1.0
     */
    @Component
    public class ConsumerTopic {
        
        //@KafkaListener(topics = {"topic01"})
        //监听topic01的0号分区和topic01的1号分区里面offset从8开始的消息。
        @KafkaListener(id = "consumer1", groupId = "felix-group", topicPartitions = {
        
                @TopicPartition(topic = "topic01", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
        })
        public void consumerTopic(ConsumerRecord<?, ?> record) {
        
            // 消费的哪个topic、partition的消息,打印出消息内容
            System.out.println("简单消费topic：" + record.topic() + "-" + record.partition() + "-" + record.value());
        }
    }

属性解释：

1.  id：消费者ID；
2.  groupId：消费组ID；
3.  topics：监听的topic，可监听多个；
4.  topicPartitions：可配置更加详细的监听信息，可指定topic、parition、offset监听。

注意：topics和topicPartitions不能同时使用；

## 2、批量消费 ##

设置application.prpertise开启批量消费即可，

# 设置批量消费
    spring.kafka.listener.type=batch
    # 批量消费每次最多消费多少条消息
    spring.kafka.consumer.max-poll-records=50

@KafkaListener(id = "consumer2",groupId = "felix-group", topics = "topic1")
    public void batchConsumer(List<ConsumerRecord<?, ?>> records) {
        
        System.out.println(">>>批量消费一次，records.size()="+records.size());
        for (ConsumerRecord<?, ?> record : records) {
        
            System.out.println(record.value());
        }
    }

## 3:ConsumerAwareListenerErrorHandler 异常处理器 ##

通过异常处理器，我们可以处理consumer在消费时发生的异常。

新建一个 ConsumerAwareListenerErrorHandler 类型的异常处理方法，用@Bean注入，BeanName默认就是方法名，然后我们将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面，当监听抛出异常的时候，则会自动调用异常处理器，

// 新建一个异常处理器，用@Bean注入
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        
        return (message, exception, consumer) -> {
        
            System.out.println("消费异常："+message.getPayload());
            return null;
        };
    }
    
    // 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
    @KafkaListener(topics = {
        "topic1"},errorHandler = "consumerAwareErrorHandler")
    public void consumerErrorHandler(ConsumerRecord<?, ?> record) throws Exception {
        
        throw new Exception("简单消费-模拟异常");
    }
    
    // 批量消费也一样，异常处理器的message.getPayload()也可以拿到各条消息的信息
    @KafkaListener(topics = "topic1",errorHandler="consumerAwareErrorHandler")
    public void batchconsumerErrorHandler(List<ConsumerRecord<?, ?>> records) throws Exception {
        
        System.out.println("批量消费一次...");
        throw new Exception("批量消费-模拟异常");
    }

## 4:消费者&生产者–消息转发 ##

我们来设想一个场景：  
双十一的时候  
1：业务系统将订单消息放入了topicA中，  
2：仓库系统将订单从A中拿到并进行业务处理，然后将处理成功的订单消息发送给topicB  
3：消息通知系统将订单消息从B中拿到后，给用户发送消息；

那么在以上业务场景中，一旦消息系统处理失败，俺么仓库系统也应该回滚，将订单在topic中的订单消息设为未处理，否则将会造成重复消费；

上述场景被称为消费者$生产者事务。

@KafkaListener(topics = {
        "topic1"})
    @SendTo("topic2")
    public String consumerToTopic(ConsumerRecord<?, ?> record) {
        
        return record.value()+"-forward message";
    }

## 5:消息拦截器 ##

拦截器可以对发送的消息进行处理，也可以对返回的消息ack等进行处理

@Component
    public class KafkaConsumer {
        
        @Autowired
        ConsumerFactory consumerFactory;
    
        // 消息过滤器
        @Bean
        public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
        
            ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
            factory.setConsumerFactory(consumerFactory);
            // 被过滤的消息将被丢弃
            factory.setAckDiscarded(true);
            // 消息过滤策略
            factory.setRecordFilterStrategy(consumerRecord -> {
        
                if (Integer.parseInt(consumerRecord.value().toString()) % 2 == 0) {
        
                    return false;
                }
                //返回true消息则被过滤
                return true;
            });
            return factory;
        }
    
        // 消息过滤监听
        @KafkaListener(topics = {
        "topic1"},containerFactory = "filterContainerFactory")
        public void onMessage6(ConsumerRecord<?, ?> record) {
        
            System.out.println(record.value());
        }
    }

## 6、定时启动、停止监听器 ##

默认情况下，当消费者项目启动的时候，监听器就开始工作，监听消费发送到指定topic的消息，那如果我们不想让监听器立即工作，想让它在我们指定的时间点开始工作，或者在我们指定的时间点停止工作，该怎么处理呢——使用KafkaListenerEndpointRegistry，下面我们就来实现：

① 禁止监听器自启动；

② 创建两个定时任务，一个用来在指定时间点启动定时器，另一个在指定时间点停止定时器；

新建一个定时任务类，用注解@EnableScheduling声明，KafkaListenerEndpointRegistry 在SpringIO中已经被注册为Bean，直接注入，设置禁止KafkaListener自启动，

@EnableScheduling
    @Component
    public class CronTimer {
        
    
        /**
         * @KafkaListener注解所标注的方法并不会在IOC容器中被注册为Bean，
         * 而是会被注册在KafkaListenerEndpointRegistry中，
         * 而KafkaListenerEndpointRegistry在SpringIOC中已经被注册为Bean
         **/
        @Autowired
        private KafkaListenerEndpointRegistry registry;
    
        @Autowired
        private ConsumerFactory consumerFactory;
    
        // 监听器容器工厂(设置禁止KafkaListener自启动)
        @Bean
        public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {
        
            ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();
            container.setConsumerFactory(consumerFactory);
            //禁止KafkaListener自启动
            container.setAutoStartup(false);
            return container;
        }
    
        // 监听器
        @KafkaListener(id="timingConsumer",topics = "topic1",containerFactory = "delayContainerFactory")
        public void onMessage1(ConsumerRecord<?, ?> record){
        
            System.out.println("消费成功："+record.topic()+"-"+record.partition()+"-"+record.value());
        }
    
        // 定时启动监听器
        @Scheduled(cron = "0 42 11 * * ? ")
        public void startListener() {
        
            System.out.println("启动监听器...");
            // "timingConsumer"是@KafkaListener注解后面设置的监听器ID,标识这个监听器
            if (!registry.getListenerContainer("timingConsumer").isRunning()) {
        
                registry.getListenerContainer("timingConsumer").start();
            }
            //registry.getListenerContainer("timingConsumer").resume();
        }
    
        // 定时停止监听器
        @Scheduled(cron = "0 45 11 * * ? ")
        public void shutDownListener() {
        
            System.out.println("关闭监听器...");
            registry.getListenerContainer("timingConsumer").pause();
        }
    }

[dbecd0f53bf64139bb6ee0b26db4590f.png]: https://img-blog.csdnimg.cn/dbecd0f53bf64139bb6ee0b26db4590f.png
[e2d155672c9e4cbf8b236df08f8edc1f.png]: https://img-blog.csdnimg.cn/e2d155672c9e4cbf8b236df08f8edc1f.png