【Kafka】基本使用：关于Consumer的几个概念

爱被打了一巴掌 2022-12-21 04:54 92阅读 0赞

上一篇，我们在[【Kafka】基本使用：Java操作Kafka（kafka-clients）][Kafka_Java_Kafka_kafka-clients]看到了使用`kafka-clients`去操作Kafka，但是在Consumer的那几个配置含义还不清楚，这篇我们就一起来看一下…

properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-gid"); // 要加入的group
    properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); // 超时，心跳
    properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 自动提交（批量）
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 新group消费为位置

### 1.enable.auto.commit ###

![在这里插入图片描述][20201114172353371.png_pic_center]

消息提交主要有以下两种方式：

*  自动提交：enable.auto.commit 可以配置自动提交时间
    
     *  可配合 auto.commit.interval.ms 设置自动提交频率
     *  可通过 enable.auto.commit=false关闭自动提交
        
        重复消费问题：设置自动提交后，若在提交间隔Consumer挂了，就可能出现已经被消费了的消息由于未被确认而再次被消费；
 *  手动提交：即使存在重复提交，也可以有有效控制重复消息数
    
     *  异步提交：consumer.commitAsync()
     *  同步提交：consumer.commitSync()，存在阻塞，因此一般批量提交，以减少阻塞频率

public void run() {
        
        consumer.subscribe(Collections.singleton(this.topic));
        List<ConsumerRecord> batchSize = new ArrayList<>();
        while (true) {
        
            // 接收消息
            ConsumerRecords<Integer, String> consumerRecords = 
                consumer.poll(Duration.ofSeconds(1));
            consumerRecords.forEach(record -> {
        
                System.out.println(record.key() + "->" + 
                                                 record.value() + "->" + record.offset());
                batchSize.add(record); // 缓存当前消息（不提交）
            }); 
            if (batch.size() >= 200) {
         // 等累积到200条消息了
                consumer.commitSync(); // 手动提交（同步提交）
                batchSize.clear();  // 清空消息缓存
            }
        }
    }

### 2.group.id ###

consumer group是kafka提供的可扩展且具有容错性的消费者机制。规则如下：

*  一个组内可以有多个消费者或消费者实例(consumer instance)，它们共享一个公共的ID，即group ID。
 *  组内的所有消费者协调在一起来消费订阅主题(subscribed topics)的所有分区(partition)。
 *  组内不能重复消费同一消息

> kafka没有像rabbitmq分成direct（单发）、fanout（广播）、topic（订阅），kafka只有topic订阅这一种模式，但他却能借助group ID实现不同功能

*  单发模式：一条消息，只能被一个group中的一个consumer消费  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]
 *  广播模式：一条消息可以被多个group消费  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]

注：若group2是group1消费一段时间后才加入的，那么它所能消费的消息由auto.oﬀset.reset决定

### 3.auto.oﬀset.reset ###

这个参数是针对新groupid中的消费者而言的，用于新groupid的消费者来消费指定的topic时，指定开始消费位置

> 由于Kafka并不是consumer消费完后，就删除相应的消息，而是根据具体的broker清理策略进行消息清理。这也意味着一条消息可以被不同consumer group的消费者重复多次消费。所以，后来的consumer group中的消费者既可以把指定topic下的所有消息都消费一遍，也可以接着前一个group消费完的位置消费。

*  auto.oﬀset.reset=latest
    
     *  新的group将会从其他消费者最后消费的oﬀset处开始消费Topic的消息
     *  group2或从group1刚消费完的消息开始消费
 *  auto.oﬀset.reset=earliest
    
     *  新的group会从该topic最早的消息开始消费
     *  group2能收到前topic所有消息，即使已经被group1消费过的消息
 *  auto.oﬀset.reset=none，新的group加入以后，由于之前不存在oﬀset，则会直接抛出异常。

### 4.max.poll.records ###

此设置限制每次调用poll返回的消息数，这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔

[Kafka_Java_Kafka_kafka-clients]: https://blog.csdn.net/weixin_43935927/article/details/109692857
[20201114172353371.png_pic_center]: /images/20221120/60e3b60344a44e61a61967e19bab5670.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/696b179b14c04c4d85d05c817089ddcf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201114200025518.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw==,size_16,color_FFFFFF,t_70#pic_center