本文主要是介绍Kafka Consumer如何实现exactly once/at least once,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
目录
- 消费端幂等性
- 消费时出现几种异常情况
- 自动提交
- 手动提交
- 精确一次消费实现
- 总结
- 至少消费一次
消费端幂等性
kafka具有两种提交offset(消费偏移量)方式,在Kafka每个分区具备一offset记录消费位置,如果消费者一直处于正常的运行转态,那么offset将没有什么用处,因为正常消费时,consumer记录了本次消费的offset和下一次将要进行poll数据的offset起始位置,但是如果消费者发生崩溃或者有新的消费者加入消费者组,就会触发再均衡Rebalance,Rebalance之后,每个消费者将会分配到新的分区,而消费者对于新的分区应该从哪里进行起始消费,这时候提交的offset信息就起作用了,提交的offset信息包括消费者组所有分区的消费进度,这时候消费者可以根据消费进度继续消费,offset自动提交是最不具确定性的,所以要使用手动提交来控制offset
消费时出现几种异常情况
自动提交
- 重复消费:当数据已经被处理,然后自动提交offset时消费者出现故障或者有新消费者加入组导致再均衡,这时候offset提交失败,导致这批已经处理的数据的信息没有记录,后续会重复消费一次
- 丢失数据:如果业务处理时间较长一点,这时候数据处理业务还未完成,offset信息已经提交了,但是在后续处理数据过程中程序发生了崩溃,导致这批数据未正常消费,这时候offset已经提交,消费者后续将不在消费这批数据,导致这批数据将会丢失
手动提交
- 重复消费(最少一次消费语义实现):消费数据处理业务完成后进行offset提交,可以保证数据最少一次消费,因为在提交offset的过程中可能出现提交失败的情况,导致数据重复消费
- 丢失数据(最多一次消费语义实现):在消费数据业务处理前进行offset提交,可以保证最多一次消费,在后续数据业务处理程序出现故障,将导致数据丢失
精确一次消费实现
offset手动提交,业务逻辑成功处理后,提交offset
解决方案:
- 对于流程中的消息,每条消息中包含唯一id,比如业务id,在数据库中将业务id作为Unique key,插入重复时会报duplicate key异常,不会导致数据库中出现脏数据
- Redis中使用set存储业务id,天然幂等性
- 如果不是上面两个场景,需要让生产者发送每条数据的时候,里面加一个全局唯一的 id,然后你这里消费到了之后,先根据这个 id 去比如 Redis 里查一下消费过吗?如果没有消费过,就处理,然后这个 id 写 Redis。如果消费过了,那就别处理了,保证不重复处理相同的消息即可
自行控制offset的准确性,这里数据不具备状态,存储使用关系型数据库,比如MySQL。这里简单说明一下实现思路:
1)利用consumer api的seek方法可以指定offset进行消费,在启动消费者时查询数据库中记录的offset信息,如果是第一次启动,那么数据库中将没有offset信息,需要进行消费的元数据插入,然后从offset=0开始消费
2)关系型数据库具备事务的特性,当数据入库时,同时也将offset信息更新,借用关系型数据库事务的特性保证数据入库和修改offset记录这两个操作是在同一个事务中进行
3)使用ConsumerRebalanceListener来完成在分配分区时和Relalance时作出相应的处理逻辑
记录kafka信息表设计
create table kafka_info(topic_group_partition varchar(32) primary key, //主题+组名+分区号 这里冗余设计方便通过这个主键进行更新提升效率 topic_group varchar(30), //主题和组名partition_num tinyint,//分区号offsets bigint default 0 //offset信息
);
代码实现
import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Duration;
import java.util.*;
public class AccurateConsumer {private static final Properties props = new Properties();private static final String GROUP_ID = "Test";static {props.put("bootstrap.servers", "192.168.142.139:9092");props.put("group.id", GROUP_ID);props.put("enable.auto.commit", false);//注意这里设置为手动提交方式props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");}final KafkaConsumer<String, String> consumer;//用于记录每次消费时每个partition的最新offsetprivate Map<TopicPartition, Long> partitionOffsetMap;//用于缓存接受消息,然后进行批量入库private List<Message> list;private volatile boolean isRunning = true;private final String topicName;private final String topicNameAndGroupId;public AccurateConsumer(String topicName) {this.topicName = topicName;topicNameAndGroupId = topicName + "_" + GROUP_ID;consumer = new KafkaConsumer<>(props);consumer.subscribe(Arrays.asList(topicName), new HandleRebalance());list = new ArrayList<>(100);partitionOffsetMap = new HashMap<>();}//这里使用异步提交和同步提交的组合方式public void receiveMsg() {try {while (isRunning) {
这篇关于Kafka Consumer如何实现exactly once/at least once的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!