kafka - yanliang's blog

消费者与消费组

消费者（Consumer）负责订阅Kafka中的主题（Topic），并且从订阅的主题上拉取消息。

与其他一些消息中间件不同的是：在Kafka的消费理念中还有一层消费组（Consumer Group）的概念，每个消费者都有一个对应的消费组。当消息发布到主题后，只会被投递给订阅它的每个消费组中的 一个 消费者。

下面的例子中，主题中共有4个分区（Partition）：P0、P1、P2、P3。有两个消费组A和B都订阅了这个主题，消费组A中有4个消费者（C0、C1、C2和C3），消费组B中有2个消费者（C4和C5）。按照Kafka默认的规则，最后的分配结果是消费组A中的每一个消费者分配到1个分区，消费组B中的每一个消费者分配到2个分区，两个消费组之间互不影响。每个消费者只能消费所分配到的分区中的消息。

每一个分区只能被一个消费组中的一个消费者所消费。

假设某一时刻某消费组内只有一个消费者 C0，订阅了一个主题，这个主题包含 7 个分区：P0、P1、P2、P3、P4、P5、P6。也就是说，这个消费者C0订阅了7个分区。

此时消费组内又加入了一个新的消费者C1，按照既定的逻辑，需要将原来消费者C0的部分分区分配给消费者C1消费。消费者C0和C1各自负责消费所分配到的分区，彼此之间并无逻辑上的干扰。

紧接着消费组内又加入了一个新的消费者C2

消费者与消费组这种模型可以让整体的消费能力具备横向伸缩性，我们可以增加（或减少）消费者的个数来提高（或降低）整体的消费能力。对于分区数固定的情况，一味地增加消费者并不会让消费能力一直得到提升，如果消费者过多，出现了消费者的个数大于分区个数的情况，就会有消费者分配不到任何分区。假设一共有8个消费者，7个分区，那么最后的消费者C7由于分配不到任何分区而无法消费任何消息。

以上分配逻辑都是基于默认的分区分配策略进行分析的，可以通过消费者客户端参数partition.assignment.strategy 来设置消费者与订阅主题之间的分区分配策略，有关分区分配的更多细节可以参考 Kafka有哪几处地方有分区分配的概念？简述大致的过程及原理

对于消息中间件而言，一般有两种消息投递模式：

• 点对点（P2P，Point-to-Point）模式
• 发布/订阅（Pub/Sub）模式

点对点模式是基于队列的，消息生产者发送消息到队列，消息消费者从队列中接收消息。

发布订阅模式定义了如何向一个内容节点发布和订阅消息，这个内容节点称为主题（Topic），主题可以认为是消息传递的中介，消息发布者将消息发布到某个主题，而消息订阅者从主题中订阅消息。主题使得消息的订阅者和发布者互相保持独立，不需要进行接触即可保证消息的传递，发布/订阅模式在消息的一对多广播时采用。Kafka 同时支持两种消息投递模式，而这正是得益于消费者与消费组模型的契合：· 如果所有的消费者都隶属于同一个消费组，那么所有的消息都会被均衡地投递给每一个消费者，即每条消息只会被一个消费者处理，这就相当于点对点模式的应用。

• 如果所有的消费者都隶属于同一个消费组，那么所有的消息都会被均衡地投递给每一个消费者，即每条消息只会被一个消费者处理，这就相当于点对点模式的应用。
• 如果所有的消费者都隶属于不同的消费组，那么所有的消息都会被广播给所有的消费者，即每条消息会被所有的消费者处理，这就相当于发布/订阅模式的应用。

消费组是一个逻辑上的概念，它将旗下的消费者归为一类，每一个消费者只隶属于一个消费组。每一个消费组都会有一个固定的名称，消费者在进行消费前需要指定其所属消费组的名称，这个可以通过消费者客户端参数group.id来配置，默认值为空字符串。

消费者并非逻辑上的概念，它是实际的应用实例，它可以是一个线程，也可以是一个进程。同一个消费组内的消费者既可以部署在同一台机器上，也可以部署在不同的机器上。

消费者客户端

一个正常的消费逻辑需要具备以下几个步骤：

1. 配置消费者客户端参数及创建相应的消费者实例。
2. 订阅主题。
3. 拉取消息并消费。
4. 提交消费位移。
5. 关闭消费者实例。

• bootstrap.servers：用来指定连接 Kafka 集群所需的 broker 地址清单，具体内容形式为host1：port1，host2：post
• group.id：消费者隶属的消费组的名称，默认值为“”。如果设置为空，会抛 InvalidGroupIdException。一般而言，这个参数需要设置成具有一定的业务意义的名称。
• key.deserializer 和 value.deserializer：消费者从broker端获取的消息格式都是字节数组（byte[]）类型，所以需要执行相应的反序列化操作才能还原成原有的对象格式。
• client.id：用来设定KafkaConsumer对应的客户端id，默认值也为“”。如果客户端不设置，则KafkaConsumer会自动生成一个非空字符串，内容形式如“consumer-1”

kafka 提供两种方式来订阅主题：

• subscribe（）
• assign （）

一个消费者可以订阅一个或多个主题，可以使用 subscribe（）方法订阅了一个主题，对于这个方法而言，既可以以集合的形式订阅多个主题，也可以以正则表达式的形式订阅特定模式的主题。

对于消费者使用集合的方式（subscribe（Collection））来订阅主题而言，比较容易理解，订阅了什么主题就消费什么主题中的消息。如果前后两次订阅了不同的主题，那么消费者以最后一次的为准。

如果消费者采用的是正则表达式的方式（subscribe（Pattern））订阅，在之后的过程中，如果有人又创建了新的主题，并且主题的名字与正则表达式相匹配，那么这个消费者就可以消费到新添加的主题中的消息。如果应用程序需要消费多个主题，并且可以处理不同的类型，那么这种订阅方式就很有效。在Kafka 和其他系统之间进行数据复制时，这种正则表达式的方式就显得很常见。

消费者不仅可以通过KafkaConsumer.subscribe（）方法订阅主题，还可以直接订阅某些主题的特定分区，在KafkaConsumer中还提供了一个assign（）方法来实现这些功能。这个方法只接受一个参数 Collection<TopicPartition> partitions ，用来指定需要订阅的分区集合。

既然有订阅，那么就有取消订阅，取消订阅的方式有如下几种

• consumer.unsubscribe();
• consumer.subscribe(new ArrayList<>());
• consumer.assgin(new ArrayList<>());

通过 subscribe（）方法订阅主题具有消费者自动再均衡的功能，在多个消费者的情况下可以根据分区分配策略来自动分配各个消费者与分区的关系。当消费组内的消费者增加或减少时，分区分配关系会自动调整，以实现消费负载均衡及故障自动转移。而通过assign（）方法订阅分区时，是不具备消费者自动均衡的功能的，其实这一点从assign（）方法的参数中就可以看出端倪，两种类型的subscribe（）都有ConsumerRebalanceListener类型参数的方法，而assign（）方法却没有。

再均衡监听器

再均衡是指分区的所属权从一个消费者转移到另一消费者的行为，它为消费组具备高可用性和伸缩性提供保障，使我们可以既方便又安全地删除消费组内的消费者或往消费组内添加消费者。不过在再均衡发生期间，消费组内的消费者是无法读取消息的。也就是说，在再均衡发生期间的这一小段时间内，消费组会变得不可用。另外，当一个分区被重新分配给另一个消费者时，消费者当前的状态也会丢失。比如消费者消费完某个分区中的一部分消息时还没有来得及提交消费位移就发生了再均衡操作，之后这个分区又被分配给了消费组内的另一个消费者，原来被消费完的那部分消息又被重新消费一遍，也就是发生了重复消费。一般情况下，应尽量避免不必要的再均衡的发生。

在subscribe（Collection＜String＞ topics，ConsumerRebalanceListener listener） 和 subscribe（Pattern pattern，ConsumerRebalanceListener listener）方法中可以指定再均衡监听器ConsumerRebalanceListener。再均衡监听器用来设定发生再均衡动作前后的一些准备或收尾的动作。ConsumerRebalanceListener 是一个接口，包含2 个方法，

void onPartitionsRevoked(Collection＜TopicPartition＞partitions)
void onPartitionsAssigned(Collection＜TopicPartition＞partitions)

onPartitionsRevoked 方法会在再均衡开始之前和消费者停止读取消息之后被调用。可以通过这个回调方法来处理消费位移的提交，以此来避免一些不必要的重复消费现象的发生。参数partitions表示再均衡前所分配到的分区。

onPartitionsAssigned方法会在重新分配分区之后和消费者开始读取消费之前被调用。参数partitions表示再均衡后所分配到的分区。

Kafka中的消费是基于拉模式的。 消息的消费一般有两种模式：推模式和拉模式。推模式是服务端主动将消息推送给消费者，而拉模式是消费者主动向服务端发起请求来拉取消息。

Kafka中的消息消费是一个不断轮询的过程，消费者所要做的就是重复地调用poll（）方法，而poll（）方法返回的是所订阅的主题（分区）上的一组消息。

对于poll（）方法而言，如果某些分区中没有可供消费的消息，那么此分区对应的消息拉取的结果就为空；如果订阅的所有分区中都没有可供消费的消息，那么poll（）方法返回为空的消息集合。

poll（）方法里还有一个超时时间参数timeout，用来控制poll（）方法的阻塞时间，在消费者的缓冲区里没有可用数据时会发生阻塞。

timeout的设置取决于应用程序对响应速度的要求，比如需要在多长时间内将控制权移交给执行轮询的应用线程。可以直接将timeout设置为0，这样poll（）方法会立刻返回，而不管是否已经拉取到了消息。如果应用线程唯一的工作就是从Kafka中拉取并消费消息，则可以将这个参数设置为最大值Long.MAX_VALUE。

poll () 方法的返回值为 ConsumerRecords （每条消息为ConsumerRecord）。在 ConsumerRecords 类中还提供了几个方法来方便开发人员对消息集进行处理：count（）方法用来计算出消息集中的消息个数，返回类型是int；isEmpty（）方法用来判断消息集是否为空，返回类型是boolean；empty（）方法用来获取一个空的消息集，返回类型是ConsumerRecord＜K，V＞。

对于Kafka中的分区而言，它的每条消息都有唯一的offset，用来表示消息在分区中对应的位置。对于消费者而言，它也有一个offset的概念，消费者使用offset来表示消费到分区中某个消息所在的位置。

在每次调用poll（）方法时，它返回的是还没有被消费过的消息集（当然这个前提是消息已经存储在Kafka 中了，并且暂不考虑异常情况的发生），要做到这一点，就需要记录上一次消费时的消费位移。并且这个消费位移必须做持久化保存，而不是单单保存在内存中，否则消费者重启之后就无法知晓之前的消费位移。再考虑一种情况，当有新的消费者加入时，那么必然会有再均衡的动作，对于同一分区而言，它可能在再均衡动作之后分配给新的消费者，如果不持久化保存消费位移，那么这个新的消费者也无法知晓之前的消费位移。

在旧消费者客户端中，消费位移是存储在ZooKeeper中的。而在新消费者客户端中，消费位移存储在Kafka内部的主题__consumer_offsets中。这里把将消费位移存储起来（持久化）的动作称为“提交”，消费者在消费完消息之后需要执行消费位移的提交。

如下图，x表示某一次拉取操作中此分区消息的最大偏移量，假设当前消费者已经消费了 x 位置的消息，那么我们就可以说消费者的消费位移为 x，图中也用了lastConsumedOffset这个单词来标识它。

不过需要非常明确的是，当前消费者需要提交的消费位移并不是 x，而是 x+1。 对应于图中的position，它表示下一条需要拉取的消息的位置。

在消费者中还有一个committed offset的概念，它表示已经提交过的消费位移。

KafkaConsumer 类提供了 position（TopicPartition）和 committed（TopicPartition）两个方法来分别获取上面所说的position和committed offset的值。

public long position(TopicPartition partition)
public OffsetAndMetadata committed(TopicPartition partition)

下面来做个小实验，验证下lastConsumedOffset、committed offset和position之间的关系。

TopicPartition tp = new TopicPartition(topic, 0);
consumer.assign(Arrays.asList(tp));
long lastConsumedOffset = -1; // 当前消费到的位移
while(true) {
  ConsumerRecords<String, String> records = consumer.poll(1000);

  if (records.isEmpty()) {
    break;
  }
  List<consumerRecord<string, string>> partitionRecords = records.records(tp);
  lastConsumedoffset = partitionRecords.get(partitionRecords.size() - 1).offset();
  consumer.commitsync(); //同步提交消费位移
  System.out.println(comsumed offset is"+ lastconsumedoffset):
  offsetAndMetadata offsetAndMetadata = consumer.committed(tp);
  System.out.println("commited offset is " + offsetAndMetadata.offset());
  long posititon = consumer.position(tp);
  System.out.println("the offset of the next record is " + posititon);
}

Output:

consumed offset is 377
commited offset is 378
the offset of the next record is 378

可以看出，消费者消费到此分区消息的最大偏移量为377，对应的消费位移lastConsumedOffset也就是377。在消费完之后就执行同步提交，但是最终结果显示所提交的位移committed offset为 378，并且下一次所要拉取的消息的起始偏移量 position 也为 378。

对于位移提交的具体时机的把握也很有讲究，有可能会造成重复消费和消息丢失的现象。

当前一次poll（）操作所拉取的消息集为[x+2，x+7]，x+2代表上一次提交的消费位移，说明已经完成了x+1之前（包括x+1在内）的所有消息的消费，x+5表示当前正在处理的位置。如果拉取到消息之后就进行了位移提交，即提交了x+8，那么当前消费x+5的时候遇到了异常，在故障恢复之后，我们重新拉取的消息是从x+8开始的。也就是说，x+5至x+7之间的消息并未能被消费，如此便发生了消息丢失的现象。

再考虑另外一种情形，位移提交的动作是在消费完所有拉取到的消息之后才执行的，那么当消费x+5的时候遇到了异常，在故障恢复之后，我们重新拉取的消息是从x+2开始的。也就是说，x+2至x+4之间的消息又重新消费了一遍，故而又发生了重复消费的现象。

在 Kafka 中默认的消费位移的提交方式是自动提交，这个由消费者客户端参数enable.auto.commit 配置，默认值为 true。当然这个默认的自动提交不是每消费一条消息就提交一次，而是定期提交，这个定期的周期时间由客户端参数auto.commit.interval.ms配置，默认值为5秒，此参数生效的前提是enable.auto.commit参数为true。

在默认的方式下，消费者每隔5秒会将拉取到的每个分区中最大的消息位移进行提交。自动位移提交的动作是在poll（）方法的逻辑里完成的，在每次真正向服务端发起拉取请求之前会检查是否可以进行位移提交，如果可以，那么就会提交上一次轮询的位移。

自动提交消费位移的方式非常简便，它免去了复杂的位移提交逻辑，让编码更简洁。但随之而来的是重复消费和消息丢失的问题。假设刚刚提交完一次消费位移，然后拉取一批消息进行消费，在下一次自动提交消费位移之前，消费者崩溃了，那么又得从上一次位移提交的地方重新开始消费，这样便发生了重复消费的现象（对于再均衡的情况同样适用）。我们可以通过减小位移提交的时间间隔来减小重复消息的窗口大小，但这样并不能避免重复消费的发送，而且也会使位移提交更加频繁。

在Kafka中还提供了手动位移提交的方式，这样可以使得开发人员对消费位移的管理控制更加灵活。很多时候并不是说拉取到消息就算消费完成，而是需要将消息写入数据库、写入本地缓存，或者是更加复杂的业务处理。在这些场景下，所有的业务处理完成才能认为消息被成功消费，手动的提交方式可以让开发人员根据程序的逻辑在合适的地方进行位移提交。开启手动提交功能的前提是消费者客户端参数enable.auto.commit配置为false。

手动提交可以细分为同步提交和异步提交，对应于 KafkaConsumer 中的commitSync（）和commitAsync（）两种类型的方法。

指定位移消费

在 Kafka 中每当消费者查找不到所记录的消费位移时，就会根据消费者客户端参数auto.offset.reset的配置来决定从何处开始进行消费，这个参数的默认值为“latest”，表示从分区末尾开始消费消息

如果将auto.offset.reset参数配置为“earliest”，那么消费者会从起始处，也就是0开始消费。

auto.offset.reset参数还有一个可配置的值—“none”，配置为此值就意味着出现查到不到消费位移的时候，既不从最新的消息位置处开始消费，也不从最早的消息位置处开始消费，此时会报出NoOffsetForPartitionException异常

如果能够找到消费位移，那么配置为“none”不会出现任何异常。如果配置的不是“latest”、“earliest”和“none”，则会报出ConfigException异常

seek() 方法：

提供的auto.offset.reset 参数也只能在找不到消费位移或位移越界的情况下粗粒度地从开头或末尾开始消费。有些时候，我们需要一种更细粒度的掌控，可以让我们从特定的位移处开始拉取消息，而 KafkaConsumer 中的 seek（）方法正好提供了这个功能，让我们得以追前消费或回溯消费。

public void seek(TopicPartition partition, long offset);

seek（）方法中的参数partition表示分区，而offset参数用来指定从分区的哪个位置开始消费。seek（）方法只能重置消费者分配到的分区的消费位置，而分区的分配是在 poll（）方法的调用过程中实现的。也就是说，在执行seek（）方法之前需要先执行一次poll（）方法，等到分配到分区之后才可以重置消费位置。

KafkaConsumer 提供了对消费速度进行控制的方法，在有些应用场景下我们可能需要暂停某些分区的消费而先消费其他分区，当达到一定条件时再恢复这些分区的消费。KafkaConsumer中使用pause（）和resume（）方法来分别实现暂停某些分区在拉取操作时返回数据给客户端和恢复某些分区向客户端返回数据的操作。

KafkaConsumer还提供了一个无参的paused（）方法来返回被暂停的分区集合。

消费者拦截器

消费者拦截器主要在消费到消息或在提交消费位移时进行一些定制化的操作。

与生产者拦截器对应的，消费者拦截器需要自定义实现org.apache.kafka.clients.consumer.ConsumerInterceptor接口。ConsumerInterceptor接口包含3个方法：

public ConsumerRecords＜K,V＞ onConsume(ConsumerRecords＜K,V＞records);
public void onCommit(Map＜TopicPartition,OffsetAndMetadata＞offsets);
public void close().

KafkaConsumer会在poll（）方法返回之前调用拦截器的onConsume（）方法来对消息进行相应的定制化操作，比如修改返回的消息内容、按照某种规则过滤消息（可能会减少poll（）方法返回的消息的个数）。如果 onConsume（）方法中抛出异常，那么会被捕获并记录到日志中，但是异常不会再向上传递。

KafkaConsumer会在提交完消费位移之后调用拦截器的onCommit（）方法，可以使用这个方法来记录跟踪所提交的位移信息，比如当消费者使用commitSync的无参方法时，我们不知道提交的消费位移的具体细节，而使用拦截器的onCommit（）方法却可以做到这一点。

在消费者中也有拦截链的概念，和生产者的拦截链一样，也是按照interceptor.classes参数配置的拦截器的顺序来一一执行的（配置的时候，各个拦截器之间使用逗号隔开）。同样也要提防“副作用”的发生。如果在拦截链中某个拦截器执行失败，那么下一个拦截器会接着从上一个执行成功的拦截器继续执行。

多线程实现

KafkaProducer是线程安全的，然而KafkaConsumer却是非线程安全的。KafkaConsumer中定义了一个 acquire（）方法，用来检测当前是否只有一个线程在操作，若有其他线程正在操作则会抛出ConcurrentModifcationException异常

KafkaConsumer中的每个公用方法在执行所要执行的动作之前都会调用这个acquire（）方法，只有wakeup（）方法是个例外

private final AtomicLong currentThread = new AtomicLong(NO_CURRENT_THREAD); // kafkaConsumer 中的成员变量

private void acquire() {
  long threadId = Thread.currentThread().getId();

  if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId)){
    throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access");
  }
  refcount.incrementAndGet();
}

acquire（）方法和我们通常所说的锁（synchronized、Lock等）不同，它不会造成阻塞等待，我们可以将其看作一个轻量级锁，它仅通过线程操作计数标记的方式来检测线程是否发生了并发操作，以此保证只有一个线程在操作。acquire（）方法和release（）方法成对出现，表示相应的加锁和解锁操作。

private void release(){
  if (refcount.decrementAndGet() == 0) {
    currentThread.set(NO_CURRENT_THREAD);
  }
}

acquire（）方法和release（）方法都是私有方法，因此在实际应用中不需要我们显式地调用，但了解其内部的机理之后可以促使我们正确、有效地编写相应的程序逻辑。

KafkaConsumer 非线程安全并不意味着我们在消费消息的时候只能以单线程的方式执行。

如果生产者发送消息的速度大于消费者处理消息的速度，那么就会有越来越多的消息得不到及时的消费，造成了一定的延迟。除此之外，由于Kafka 中消息保留机制的作用，有些消息有可能在被消费之前就被清理了，从而造成消息的丢失。我们可以通过多线程的方式来实现消息消费，多线程的目的就是为了提高整体的消费能力。

多线程的实现方式大致有如下三种方式：

第一种也是最常见的方式：线程封闭，即为每个线程实例化一个KafkaConsumer对象

一个线程对应一个KafkaConsumer实例，我们可以称之为消费线程。一个消费线程可以消费一个或多个分区中的消息，所有的消费线程都隶属于同一个消费组。这种实现方式的并发度受限于分区的实际个数。

内部类KafkaConsumerThread代表消费线程，其内部包裹着一个独立的KafkaConsumer实例。通过外部类的main（）方法来启动多个消费线程，消费线程的数量由consumerThreadNum变量指定。一般一个主题的分区数事先可以知晓，可以将consumerThreadNum设置成不大于分区数的值，如果不知道主题的分区数，那么也可以通过KafkaConsumer类的partitionsFor（）方法来间接获取，进而再设置合理的consumerThreadNum值。

上面这种多线程的实现方式和开启多个消费进程的方式没有本质上的区别，它的优点是每个线程可以按顺序消费各个分区中的消息。缺点也很明显，每个消费线程都要维护一个独立的TCP连接，如果分区数和consumerThreadNum的值都很大，那么会造成不小的系统开销。

指定分区消费

第一种方案中，由于消费者与分区数的关系，当消费线程的个数大于分区数时，就有部分消费线程一直处于空闲的状态。

第二种方案是多个消费线程同时消费同一个分区，这个通过 assign（）、seek（）等方法实现，这样可以打破原有的消费线程的个数不能超过分区数的限制，进一步提高了消费的能力。不过这种实现方式对于位移提交和顺序控制的处理就会变得非常复杂，实际应用中使用得极少，并不推荐。

一般而言，分区是消费线程的最小划分单位。

改造消息处理模块

在第一种方案的具体实现的第①行，如果这里对消息的处理非常迅速，那么 poll（）拉取的频次也会更高，进而整体消费的性能也会提升；相反，如果在这里对消息的处理缓慢，比如进行一个事务性操作，或者等待一个RPC的同步响应，那么poll（）拉取的频次也会随之下降，进而造成整体消费性能的下降。一般而言，poll（）拉取消息的速度是相当快的，而整体消费的瓶颈也正是在处理消息这一块，如果我们通过一定的方式来改进这一部分，那么我们就能带动整体消费性能的提升。

第三种方案，可以尝试将处理消息模块改成多线程的实现方式，来提升性能。

第三种实现方式相比第一种实现方式而言，除了横向扩展的能力，还可以减少TCP连接对系统资源的消耗，不过缺点就是对于消息的顺序处理就比较困难了。