Kafka的offset自定义存储实现

太过爱你忘了你带给我的痛 2022-12-27 08:00 142阅读 0赞

# 一、什么是Offset #

在kafka中，每一条消息都有一个与之对应的序列号，这个序列号就是offset，表示消息的偏移量。

特点：

*  偏移量从0开始递增
 *  topic中的每个分区维护自己的一个offset

#    二、Offset存储位置 #

每个消费者本地内存中会存储自己当前消费消息的offset，以便下次消费消息从offset+1开始消费。

此外，消费者每次消费完消息后还要将offset提价到kafka集群，kafka集群会将offset保存到kafka本地。

对应的就是一条group\_id + partition\_id + offset记录，表示当前消费者组在某个分区的消费偏移量。

# 三、Rebalance再平衡 #

Rebalance就是当消费者组内有消费者增加或减少时，需要对分区重新进行分配，有的消费者可能会分配到新的分区，所以需要通过Rebalance来拉取订阅的topic的所有分区的偏移量信息到本地。避免同一个消费者组内发生重复消费的情况。

# 四、offset提交 #

offset必须提交，不然消费者重启后又会重头开始消费消息（除非消息已过7天有效期被清除），offset提交有以下三种方式：

## 自动提交 ##

自动提交即消费者消费完消息自动提交，不需要开发人员调用提交方法，客户端自动提交offset到kafka本地。kafka默认开启自动提交，由以下参数设置：ConsumerConfig.ENABLE\_AUTO\_COMMIT\_CONFIG；

自动提交有一个配置参数，用来设置自动提交offset的延迟时间，即消费者拉取消息后多久提交偏移量，由以下参数设置：ConsumerConfig.AUTO\_COMMIT\_INTERVAL\_MS\_CONFIG，单位毫秒；该参数生效的前提是要开启自动提交。

## 手动提交 ##

kafka也提供了开发者自动提交offset的方法，首先需要设置ConsumerConfig.ENABLE\_AUTO\_COMMIT\_CONFIG参数为false，禁用自动提交。

手动提交有同步提交和异步提交两种方式：

1.  同步提交：consumer.commitSync();  
                同步提交方法如果提交不成功，会一直重试，阻塞消费者线程，效率低，不建议使用
2.  异步提交：
    
        consumer.commitAsync((map, e) -> {
        				if (e == null) {
        					System.out.println("offset提交成功");
        				} else {
        					System.out.println("offset提交失败");
        				}
        			});
    
         异步提交不会阻塞消费者线程，采用异步方式，会传入一个回调函数，来接受返回结果。

## 自定义存储 ##

不管是同步提交还是异步提交，offset都是保存到kafka本地，offset提交操作和业务操作不是一个事务，就有可能出现问题：

1.  业务操作成功，offset提交失败：会造成重复消费
2.  业务操作失败，offset提交成功：会造成消费者丢数据

因此，kafka也支持自定义存储offset，将offset提交和业务操作放在一个事务中，就能避免出现这两种情况。自定义存储也需要关闭kafka的自动提交，自定义存储offset需要完成以下三件事：

1.  消费者维护一个Map，保存自己当前的消费offser信息
2.  消费者消费消息后将offset提交保存到自定义的存储
3.  Rebalance时消费者拉取offset，避免重复消费

第一、二步开发者自己实现offset的存储逻辑即可，第三步kafka提供了一个ConsumerRebalanceListener监听器用来监听Rebalance事件，开发者需要实现这个RebalanceListener的两个方法：

1.  onPartitionsRevoked方法：在Rebalance之前调用，将当前消费者内存中保存的offset信息保存到自定义的存储中，即将消费者维护的Map序列化到数据库；
2.  onPartitionsAssigned方法：在Rebalance之后调用，清空消费者内存中的offset信息和重新读取自定义存储汇总保存的offset定义，定位到最近提交的offset位置继续消费。

# 五、自定义存储Offset到Mysql中的实现 #

## 自定义RebalanceListener ##

package cn.lsh.kafka.listener;
    
    import cn.lsh.kafka.db.entity.OffsetEntity;
    import cn.lsh.kafka.db.service.OffsetService;
    import cn.lsh.kafka.db.service.ServiceFactory;
    import org.apache.kafka.clients.consumer.Consumer;
    import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
    import org.apache.kafka.common.TopicPartition;
    
    import java.util.Collection;
    import java.util.Date;
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * 自定义存储offset时，实现自己的Rebalance监听器
     */
    public class MyConsumerRebalanceListener implements ConsumerRebalanceListener {
    	private Consumer consumer;
    
    	public static final Map<TopicPartition, Long> currentOffset = new HashMap<>();
    
    	public MyConsumerRebalanceListener(Consumer consumer) {
    		this.consumer = consumer;
    	}
    
    	@Override
    	public void onPartitionsRevoked(Collection<TopicPartition> topicPartitions) {
    		//方法在Rebalance之前调用，将当前消费者
    		// OffsetService service = new OffsetServiceImpl();
    		OffsetService service = ServiceFactory.buildService(OffsetService.class);
    		commitOffset(consumer.groupMetadata().groupId(), service);
    		// service.commit();
    		// service.colse();
    	}
    
    	@Override
    	public void onPartitionsAssigned(Collection<TopicPartition> topicPartitions) {
    		System.out.println("Rebalance拉取offset");
    		String groupId = consumer.groupMetadata().groupId();
    		//方法在Rebalance之后调用
    		currentOffset.clear();//清空客户端内存中的offset
    		for (TopicPartition partition : topicPartitions) {
    			//定位到最近提交的offset位置继续消费
    			consumer.seek(partition, getOffset(groupId, partition));
    		}
    	}
    
    	/**
    	 * 自定义实现获取offset
    	 *
    	 * @param partition
    	 * @return
    	 */
    	private static long getOffset(String groupId, TopicPartition partition) {
    		OffsetService service = ServiceFactory.buildService(OffsetService.class);
    		OffsetEntity entity = service.queryByCond(groupId, partition.topic(), partition.partition());
    		if (entity != null) {
    			return entity.getValue();
    		}
    		return 0;
    	}
    
    	/**
    	 * 自定义实现保存offset
    	 */
    	public static void commitOffset(String groupId, OffsetService service) {
    		Date now = new Date();
    		for (Map.Entry<TopicPartition, Long> entry : currentOffset.entrySet()) {
    			TopicPartition partition = entry.getKey();
    			OffsetEntity entity = service.queryByCond(groupId, partition.topic(), partition.partition());
    			//注意：消费者保存到currentOffset的offset是当前消费消息的偏移量，这里提交到mysql需要加1,不然Rebalance后会重复消费一条消息
    			long nextOffset = entry.getValue() + 1;
    			if (entity == null) {
    				OffsetEntity offset = new OffsetEntity();
    				offset.setGroupId(groupId);
    				offset.setTopic(partition.topic());
    				offset.setPartitionId(partition.partition());
    				offset.setValue(nextOffset);
    				offset.setCreateDate(now);
    				offset.setUpdateDate(now);
    				service.insert(offset);
    			} else if (nextOffset != entity.getValue()) {
    				service.update(entity.getId(), nextOffset);
    			}
    		}
    	}
    }

## 消费者代码 ##

package cn.lsh.kafka.consumer;
    
    import cn.lsh.kafka.db.service.BusinessService;
    import cn.lsh.kafka.db.service.OffsetService;
    import cn.lsh.kafka.db.service.ServiceFactory;
    import cn.lsh.kafka.db.service.impl.BusinessServiceImpl;
    import cn.lsh.kafka.listener.MyConsumerRebalanceListener;
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.TopicPartition;
    
    import java.time.Duration;
    import java.time.LocalTime;
    import java.util.Collections;
    import java.util.Properties;
    
    public class OffsetConsumer {
    
    	public static void main(String[] args) {
    		String groupId = "group_1";
    		Properties properties = new Properties();
    		properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node00:9092,node01:9092,node02:9092");
    		properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
    		properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    
    		properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    		// properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    		properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
    		properties.setProperty(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG, "true");
    
    		//关闭自动提交offset
    		properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    
    		KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(properties);
    
    		//订阅topic时设置自定义的Rebalance监听器
    		consumer.subscribe(Collections.singletonList("ddd-1"), new MyConsumerRebalanceListener(consumer));
    
    		OffsetService offsetService = ServiceFactory.buildService(OffsetService.class);
    		BusinessService busiService = new BusinessServiceImpl();
    		while (true) {
    			ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
    			if (records.count() > 0) {
    				System.out.println("当前批次拉取数据：" + records.count() + "，时间：" + LocalTime.now());
    			}
    			for (ConsumerRecord<Integer, String> record : records) {
    				System.out.println(record.offset() + "     " + record.key() + "     " + record.value());
    				TopicPartition partition = new TopicPartition(record.topic(), record.partition());
    				MyConsumerRebalanceListener.currentOffset.put(partition, record.offset());
    				//业务处理
    				busiService.deal(record.value());
    			}
    			//保存offset到mysql
    			MyConsumerRebalanceListener.commitOffset(groupId, offsetService);
    			//提交事务，让业务处理和提交offset保持一个事物
    			ServiceFactory.commit();
    		}
    	}
    }

offset存储表的实体类字段：

public class OffsetEntity implements Serializable {
        private static final long serialVersionUID = 228985855212725051L;
        
        private Integer id;
        /**
        * 消费者组
        */
        private String groupId;
        /**
        * 主题
        */
        private String topic;
        /**
        * 分区
        */
        private Integer partitionId;
        /**
        * 偏移量
        */
        private Long value;
    
        private Date createDate;
    
        private Date updateDate;
    
        ....
    }

[完整代码传送门][Link 1]

[Link 1]: https://gitee.com/jlzl/hadoop-test/tree/master/src/main/java/cn/lsh/kafka/listener