SpringBoot集成kafka-监听器手动确认接收消息（主要为了保证业务完成后再确认接收）

本文主要是介绍SpringBoot集成kafka-监听器手动确认接收消息（主要为了保证业务完成后再确认接收），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

SpringBoot集成kafka-监听器手动确认接收消息

1、说明
2、示例
- 2.1、application.yml
- 2.2、消费者
- 2.3、生产者
- 2.4、测试类
- 2.5、测试

1、说明

kafak中默认情况下是自动确认消息接收的，也就是说先启动消费者监听程序，再启动生产者发送消息，此时消费者监听到生产者发送的消息后，程序会自动确认接收成功，偏移量会自动下移，此时再启动消费者，偏移量会从新的位置读取数据，如果本次出现异常，业务没有处理完成，那么下次启动消费者是读取不到本次的消息的，所以可以采用手动确认的配置，确保本次消费者接收到了消息，并且业务正常处理完毕了，给kafak手动反馈接收成功。

在这里插入图片描述

2、示例

在这里插入图片描述

2.1、application.yml

在这里插入图片描述

2.2、消费者

package com.power.consumer;import com.power.model.User;
import com.power.util.JSONUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;import java.util.function.Consumer;@Component
public class EventConsumer {@KafkaListener(topics = {"${kafka.topic.name}"},groupId="${kafka.consumer.group}")public void onEvent4(String userJson,@Header(value=KafkaHeaders.RECEIVED_TOPIC) String topic,@Header(value=KafkaHeaders.RECEIVED_PARTITION_ID) String partition,ConsumerRecord<String,String> record,Acknowledgment ack){try {User user =JSONUtils.toBean(userJson,User.class);System.out.println("读取/消费到的事件,user："+user+",topic："+topic+",partition："+partition);System.out.println("读取/消费到的事件："+record.toString());int a = 10/0;//业务确认完成，给kafka服务器反馈确认ack.acknowledge();//手动确认消息，就是告诉kafka服务器，该消息我已经接收到了，默认情况下是自动确认//手动确认后，下次启动消费者，偏移量会从新的位置开始；没有手动确认，下次启动消费者，偏移量还是从老位置开始}catch (Exception e){e.printStackTrace();}}}

2.3、生产者

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 sendEvent2(){User user = User.builder().id(10001).phone("15676767676").birthday(new Date()).build();String userJson = JSONUtils.toJSON(user);kafkaTemplate.send("helloTopic",userJson);}}

2.4、测试类

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 SpringBoot02KafkaBaseApplication {@Resourceprivate EventProducer eventProducer;@Testvoid sendEvent2(){eventProducer.sendEvent2();}}