Kafka-2.6.0之Consumer

小灰灰 2022-12-05 05:25 161阅读 0赞

### 文章目录 ###

*  1. 偏移量和消费者位置（Offsets and Consumer Position）
 *  2. 群体消费和主题订阅（Consumer Groups and Topic Subscriptions）
 *   *  2.1 Consumer Groups为何出现
     *  2.2 Consumer Groups原理
     *  2.3 Consumer Groups组员关系
     *  2.4 保存\`records\`和\`offset\`
     *  2.4.1 方法一，手动
     *  2.4.2 方法二，自动
 *  3. 检测消费失败（Detecting Consumer Failures）
 *  4. 自动提交示例
 *  5. 手动提交示例
 *   *  5.1 批量手动提交
     *  5.2 根据偏移量提交
 *  6. 手动指定消费分区
 *   *  6.1 使用场景
     *  6.2 如何使用
 *  7. 控制consumer的position
 *  8. 读取事务消息

# 1. 偏移量和消费者位置（Offsets and Consumer Position） #

Kafka为每个`分区`（Partition）中每条记录保存一个`偏移量`（Offset ），`offset`是记录在分区中的唯一标识，也表示`消费者`（consumer）在分区中的`位置`(position)。例如：一个`offset` 在5的消费者，已经消费了0到4的记录，它的下个消费记录就是5。  
消费者位置给出了下一个偏移量的大小（在消费者现在处于分区中的位置大小加1），偏移量将在每次调用`poll(Duration)`方法时更新（当该方法返回的记录为空时，偏移量不变）。  
`commit position`是最后一个被安全保存的`offset`，如果进程宕机并且重启，`consumer`将恢复到该`offset`。`consumer`可以自动的提交`offset`，也可以选择手动提交。

# 2. 群体消费和主题订阅（Consumer Groups and Topic Subscriptions） #

## 2.1 Consumer Groups为何出现 ##

为了分担消费和处理`records`的工作，kafka使用`群体消费`（Consumer Groups）的概念，它允许一系列的进程来共同消费`records`。这些进程可以是同一个进程，也可以是分布在不同机器的多个进程。所有共享同一个`group.id`的消费者实例是同一个消费者群体。  
从概念上讲，可以将`group`看作是一个由多个进程组成的单个逻辑订阅者。作为一个`multi-subscriber`系统，Kafka自然支持在不复制数据的情况下为给定的主题分配任意数量的用户组`groups`.

## 2.2 Consumer Groups原理 ##

组中的每个consumer可以通过`subscribe`方法动态的设置一个或者多个想要订阅的`topic`，kafka将会把每个消息发送给订阅主题的每个`consumer group`中的一个进程，这是通过把所有的分区负载均衡到`consumer group`中的所有成员上，因此每个`partition`实际是对应一个`group`中的一个成员。例如：1个topic有4个`partition`，有个拥有两个进程的`consumer group`，每个进程将从两个`partition`消费消息。

## 2.3 Consumer Groups组员关系 ##

`consumer group` 之间的组员关系是动态维护的，如果一个进程宕机了，分配给它的`partition`将会被重新分配到同一个组的其它成员。同样的，一个组进入了新的`consumer`，`partition`将会从已存在的`consumers`上转移到新的`consumer`上。这是通过`rebalancing`实现的， `Group rebalancing`也用于订阅的topic加入了新的partitions或者创建一个新的topic匹配到了一个`subscribed regex`。`group`将会通过定期的刷新元数据来自动检测新的 `partitions`，将它们分配到`group`中

## 2.4 保存`records`和`offset` ##

## 2.4.1 方法一，手动 ##

1.  设置参数`enable.auto.commit=false`
2.  从`ConsumerRecord`获取所需的信息和`offset`保存起来
3.  在启动时保存信息调用该方法`seek(TopicPartition, long)`.

## 2.4.2 方法二，自动 ##

实现接口`ConsumerRebalanceListener`，创建该接口实例。调用`consumer`的 `subscribe(Collection, ConsumerRebalanceListener)`或者`subscribe(Pattern, ConsumerRebalanceListener)`方法\`

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener { 
           private Consumer<?,?> consumer;
    
           public SaveOffsetsOnRebalance(Consumer<?,?> consumer) { 
               this.consumer = consumer;
           }
    
           public void onPartitionsRevoked(Collection<TopicPartition> partitions) { 
               // save the offsets in an external store using some custom code not described here
               for(TopicPartition partition: partitions)
                  saveOffsetInExternalStore(consumer.position(partition));
           }
    
           public void onPartitionsLost(Collection<TopicPartition> partitions) { 
               // do not need to save the offsets since these partitions are probably owned by other consumers already
           }
    
           public void onPartitionsAssigned(Collection<TopicPartition> partitions) { 
               // read the offsets from an external store using some custom code not described here
               for(TopicPartition partition: partitions)
                  consumer.seek(partition, readOffsetFromExternalStore(partition));
           }
       }

# 3. 检测消费失败（Detecting Consumer Failures） #

订阅了一系列主题后，`consumer`将在调用`poll(Duration)`方法时自动加入到`group`中。在后台，`consumer`将会定期的发送心跳给`server`，如果`consumer`宕机或者在设定时间 *`session.timeout.ms`* 没有发送心跳将被认为是`死亡`，它拥有的`partition`将会重新被分配。  
也可能存在这样一种情况，`consumer`还是会继续发送心跳，但是它已经不能消费消息，它将会无期限的保留`partition`，可以使用参数 *`max.poll.interval.ms`* 设置存活检测机制，超过设定时间`consumer`一直没有拉取到信息就会主动离开`group`，以便其它`consumer`接管它的`partition`，有以下两个参数可以控制`poll loop`

1.  *`max.poll.interval.ms`*： 通过此参数增加轮询间隔，也可以给`consumer`更多的时间来处理获取到的消息，缺点是会增加一个组的重新平衡，因为`group`只有在`poll`期间才会重新平衡
2.  *`max.poll.records`*：限制每次`poll`最大`records`，便于预测每次轮询处理的消息数，能够减少轮询时间（只要达到这个数目），也可以减少`group rebalancing`的影响

# 4. 自动提交示例 #

Properties props = new Properties();
    props.setProperty("bootstrap.servers", "localhost:9092");
    props.setProperty("group.id", "test");
    props.setProperty("enable.auto.commit", "true");
    props.setProperty("auto.commit.interval.ms", "1000");
    props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Arrays.asList("foo", "bar"));
    while (true) { 
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records)
            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }

*`enable.auto.commit`*： 控制是否自动提交  
*`auto.commit.interval.ms`*：控制自动提交频率  
示例中 `consumer` 订阅了`foo`和`bar`两个主题，使用参数 *`group.id`* 注册到名为`test`的分组中

# 5. 手动提交示例 #

## 5.1 批量手动提交 ##

与依赖`consumer`定期的提交消费偏移量相反，我们可以控制`records`是否是消费从而提交`offset`，比如：有些`records`和业务逻辑联系紧密，我们就可以把它们看做一个整体去消费，直到所有消息都被完全处理。

Properties props = new Properties();
    props.setProperty("bootstrap.servers", "localhost:9092");
    props.setProperty("group.id", "test");
    props.setProperty("enable.auto.commit", "false");
    props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Arrays.asList("foo", "bar"));
    final int minBatchSize = 200;
    List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
    while (true) { 
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) { 
            buffer.add(record);
        }
        if (buffer.size() >= minBatchSize) { 
            insertIntoDb(buffer);
            consumer.commitSync();
            buffer.clear();
        }
    }

> 此示例中，当拉取的消息大于等同于200时，先进行业务逻辑处理：比如插入到db中，然后再进行提交。如果使用自动提交，提交成功后插入到db可能会失败。但是这种情况也有弊端，加入在插入db后提交之前服务宕机，那么就会丢失一系列的消息，这一系列的消息在服务重启后就会重复消费，下面示例可以更细粒度控制提交偏移量

## 5.2 根据偏移量提交 ##

try { 
        while(running) { 
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
            for (TopicPartition partition : records.partitions()) { 
                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                for (ConsumerRecord<String, String> record : partitionRecords) { 
                    System.out.println(record.offset() + ": " + record.value());
                }
                long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
            }
        }
    } finally { 
      consumer.close();
    }

Note: The committed offset should always be the offset of the next message that your application will read. Thus, when calling commitSync(offsets) you should add one to the offset of the last message processed.

> 提交的偏移量应该是下条消息的偏移量，所以当调用`commitSync(offsets)`方法时应该在最后一次的偏移量的基础上加1

# 6. 手动指定消费分区 #

## 6.1 使用场景 ##

1.  `consumer`正在维护和分区相关的一些本地状态，它就应该从该分区获取消息
2.  如果`consumer`是高可用的，并且在宕机后能自动重启，那么没必要让`kafka`进行失败检测，重新分配分区

## 6.2 如何使用 ##

String topic = "foo";
    TopicPartition partition0 = new TopicPartition(topic, 0);
    TopicPartition partition1 = new TopicPartition(topic, 1);
    consumer.assign(Arrays.asList(partition0, partition1));

> 该方式不会使用组协调机制（group coordination），因此`consumer`宕机不会rebalance，就算是`consumer`共用一个`groupid`也是相互独立的，但是为了避免commit offset 冲突应该避免使用同一个`groupid`

# 7. 控制consumer的position #

kafka允许用户手动控制postion位置，可以向前或者向后移动，kafka提供了一下4个方法

1.  `public void seek(TopicPartition partition, long offset)`
2.  `public void seek(TopicPartition partition, OffsetAndMetadata offsetAndMetadata)`
3.  `public void seekToBeginning(Collection<TopicPartition> partitions)`：找到所有partitions的第一个 offset
4.  `public void seekToEnd(Collection<TopicPartition> partitions)` ：找到所有partitions的最后一个 offset

# 8. 读取事务消息 #

consumer 需要注册该属性 *`isolation.level=read_committed`*，此模式下 consumer 只读取 producer 提交成功的事务性消息，然后和之前一样读取费事务性消息，此模式还会过滤掉`aborted`消息。  
事务性消息会拥有提交或者回滚的标记来表明他们是事务性消息。这些消息不会反馈给 consumer ，但是会有偏移量日志，因此 consumer 读取拥有事务性消息的主题的offset时可能会看到断裂。这些丢失的消息会成为事务性标记，并且被过滤掉。