Apache Kafka 0.9 KafkaConsumer

女爷i 2022-09-21 11:53 159阅读 0赞

Kafka开始被设计时，带有一个Scala的Producer和Consumer。而这些API有很多的缺陷，例如，支持Consumer Group且容错的“high-level” Consumer API，不能支持太多更复杂的情况。另一种“simple” Consumer Client提供了所有的操作，但是它要求用户自己管理故障和错误。所以，重新设计了可以应对大量旧Clients很难甚至不可能处理的用户场景。

重写Producer API的第一阶段是在0.8.1。最新的0.9完成了第二阶段，并且有新的Consumer API。基于一组全新的Kafka自己的协调协议，新Consumer有以下优点：

1.  统一的API：新Consumer结合了旧的“simple”和“high-level” Consumer Clients的功能，提供了用组织协调和低水平访问建立你自己的消费策略。
2.  减少了依赖：新Consumer用纯Java重写。它不在依赖与Scala运行期和ZooKeeper，这使得你的工程包含更轻量级的库。
3.  更安全：在Kafka 0.9实现了安全扩展（security extensions），且仅被新Consumer支持。
4.  这种新Consumer还新增了一套管理容错的Consumer处理组。以前这个功能是用一个笨重的Java Client实现（与ZooKeeper互动严重）。这种逻辑的复杂性使得难以用其他语言构建一个具有全部特性的Consumer。随着这种新协议的引入，其变得很容易实现。

虽然这种新的Consumer使用了重新设计的API和新的协调协议，但是概念上并非从根本上不同。所以熟悉旧Consumer的用户不会有一些问题需要理解。然后，在遵守Group管理和线程模型上有一个细微的细节需要额外的关注。本文的目的将包含新Consumer的基本使用和解释所有这些细节。

# 介绍 #

在开始编码前，我们先熟悉下基本概念。在Kafka中，每个topic被划分到一个叫做Partition的日志集合中。Producers以自己的速度写这些日志，Consumers以自己的速度读这些日志。Kafka通过在一个共享公共的group identifier的Consumer Group中分配Partitions来拓展topic的消费。下图是一个有三个Partition的topic和一个有两个成员的Consumer Group。该topic中的每个Partition只被精确地分配给一个Consumer。

![Center][]

虽然旧的Consumer依赖于ZooKeeper管理Group，但是新的Consumer使用一组内置在Kafka的协调协议。对于每个Group，Brokers中的某一个被选为Group Coordinator。该Coordinator负责管理对应Group的状态。它的主要任务是当新的Consumer成员加入或老的Consumer成员消失，和更改topic的metadata时，协调Partition的分配。重新分配Partition的过程称作对该Group的负载均衡。

当一个Group第一次被初始化时，在每个Partition中，Consumers要么从最早的offset读数据，要么从最晚的offset读数据。每个Partition中的消息被顺序的读取。当Consumer开始处理，它就会提交已经成功处理过的消息的offset。例如，在下图中，Consumer的位置在offset 6，它上次提交的offset是offset 1。

![Center 1][]

当一个Partition重新分配到Group中的另一个Consumer时，初始Partition被设置成最近一次提交的offset。如果上例中的Consumer突然宕了，接管该Partition的Group中Consumer将会从offset 1开始消费。那样一来，它不得不重新处理已宕的Consumer的offset 6。

该图也给出了log中另外两个重要的Partitions。Log End Offset是最后一条message写入log的offset。High Watermark是最后一条message被成功复制到所有的log副本中的offset。从Consumer的角度看，最主要的是知道你只能读到High Watermark。这将防止Consumer读没有副本可能稍后会丢失的数据。

# 配置和初始化 #

为了开始学习Consumer，你需要在你的project中添加kafka-clients依赖。Maven片段如下：

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.9.0.0</version>
    </dependency>

Consumer用Properties文件构造，就像其他的Kafka Clients一样。在下面的例子中，我提供了一个使用Consumer Groups最简单的配置：

Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("group.id", "consumer-tutorial");
    props.put("key.deserializer", StringDeserializer.class.getName());
    props.put("value.deserializer", StringDeserializer.class.getName());
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

就像在旧Consumer和Producer中一样，我们需要为Consumers配置一个Brokers的初始化列表，以便能访问集群中的其他node。但是，不需要提供集群中所有的server——Client从列表中这些Brokers中确定全部存活的Broker。这里我们已经假定该Broker是在localhost上运行。同时，Consumer也需要知道怎么反序列化message的Key和Value。最后，为了添加到一个Consumer Group中，我们需要配置一个Group Id。我们继续阅读本文，将会介绍更多的配置。

# Topic订阅 #

为了可以开始消费，你必须先订阅（subscribe）一个你的程序需要读取的Topic。在下例中，我们subscribe了foo和bar两个Topic：

consumer.subscribe(Arrays.asList("foo", "bar"));

当你subscribe之后，该Consumer将会和这个Group中的其他Consumer协调配合来分配Partition。当你开始消费数据时，这个过程自动完成。后面我们也会介绍怎么使用assign API去人工手动分配Partition，但是请切记不能同时混合使用自动和手动两种分配方式。

subscribe方法是不能增加的，你必须在列表里包含所有你想要消费的Topic。你可以在任何时候改变你已经subscribe的Topic集合——当你调用subscribe方法时，任何之前订阅的Topic都将被新的Topic列表所替代。

# 基本的Poll循环 #

Consumer需要并行地从许多Partition获取数据，因为许多Topic可能分布在许多不同的Broker中。为了实现这一点，它使用的API方式和Unix中的poll或者select：一旦Topic被注册，未来所有的协调、负载均衡、和数据获取都通过一个专门在事件循环中被调用的单独的poll方法来驱动。这就需要一个简单的、高效的可以从单个线程中操作所有IO的实现。

在subscribe Topic之后，你需要开启事件循环来得到Partition的分配，并且开始获取数据。这听起来很复杂，但是你需要做的所有事情只是在循环中调用poll方法，剩下的其他事情就由Consumer来处理。每次调用poll方法，将返回一个来自已分配的Partition的message集合（可能是空的）。下面的例子展示了一个基本的poll循环，它将打印获取到的数据的offset和记录的value：

poll API返回从当前Position获取的记录。当这个Group是第一次创建，Position将根据重置策略（reset policy，要么为每个Partition设置为最早的offset，要么为每个Partition设置为最晚的offset）设置。一旦Consumer开始提交offsets，随后每次负载均衡将重置Position为最后一次提交的offset。poll()方法的参数控制当Consumer在当前Position等待记录时，它将阻塞的最大时长。当有记录到来时，Consumer将会立即返回。但是，在返回前如果没有任何记录到来，Consumer将等待直到超出指定的等待时长。

Consumer被设计成运行在自己所属的线程中。在没有外部同步的情况下，使用多线程是不安全的。同时，它很可能不是一个好的主意去尝试。在这个例子中，我们使用一个flag，用于当程序shutdown时，跳出poll循环。当其他线程设置flag为false时（如，shutdown该进程），poll()方法一返回就跳出循环。同时，程序完成处理任何已被返回的记录。

当你使用完Consumer时，你应该总是关闭该Consumer。这样做不仅为了清理使用的Sockets，也确保Consumer告诉Coordinator要退出该Group。

这个例子使用了一个相对短的超时时长，以确保当关闭Consumer时，不会有太大的延迟。另外，你可以设置一个长的超时，同时用wakeup API跳出循环。

try {
      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
        for (ConsumerRecord<String, String> record : records)
          System.out.println(record.offset() + “: ” + record.value());
      }
    } catch (WakeupException e) {
      // ignore for shutdown
    } finally {
      consumer.close();
    }

我们改变超时为Long.MAX\_VALUE，这基本上意味着Consumer将无期限地阻塞，直到下一条记录达到。替代上一个例子中的flag，线程调用consumer.wakeup()中断活跃的poll()来触发shutdown，造成它的原因是抛出WakeupException。这个API是可以在另一个线程安全的使用。注意如果在处理过程中没有活跃的poll()，这个异常将被在下一次调用时抛出。在这个例子中，我们捕获这个异常，阻止它传播。

# 整合 #

在接下来的例子中，我们将这些都放在一起去构建一个简单的Runnable任务，它包含初始化Consumer，subscribe一个Topic列表，无限循环执行poll()直到外力地关闭。

public class ConsumerLoop implements Runnable {
      private final KafkaConsumer<String, String> consumer;
      private final List<String> topics;
      private final int id;
    
      public ConsumerLoop(int id,
                          String groupId, 
                          List<String> topics) {
        this.id = id;
        this.topics = topics;
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put(“group.id”, groupId);
        props.put(“key.deserializer”, StringDeserializer.class.getName());
        props.put(“value.deserializer”, StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<>(props);
      }
     
      @Override
      public void run() {
        try {
          consumer.subscribe(topics);
    
          while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
            for (ConsumerRecord<String, String> record : records) {
              Map<String, Object> data = new HashMap<>();
              data.put("partition", record.partition());
              data.put("offset", record.offset());
              data.put("value", record.value());
              System.out.println(this.id + ": " + data);
            }
          }
        } catch (WakeupException e) {
          // ignore for shutdown 
        } finally {
          consumer.close();
        }
      }
    
      public void shutdown() {
        consumer.wakeup();
      }
    }

为了测试这个例子，你需要一个运行的0.9.0.0的Kafka Broker，和一个有string类型数据可以消费的Topic。写大量string类型数据到Topic的最简单的方法是使用kafka-verifiable-producer.sh脚本。为了更有意思，我们需要使这个Topic有多个Partition，这样就不会有一个Consumer去做所有事情。例如，一个Kafka Broker和ZooKeeper都运行在localhost，你可以在Kafka的根目录像下面一样操作：

# bin/kafka-topics.sh --create --topic consumer-tutorial --replication-factor 1 --partitions 3 --zookeeper localhost:2181
    # bin/kafka-verifiable-producer.sh --topic consumer-tutorial --max-messages 200000 --broker-list localhost:9092

然后，我们创建一个简单的驱动程序来启动有3个Consumer的Consumer Group，所有的Consumer都subscribe我们已经创建的相同的Topic。

public static void main(String[] args) { 
      int numConsumers = 3;
      String groupId = "consumer-tutorial-group"
      List<String> topics = Arrays.asList("consumer-tutorial");
      ExecutorService executor = Executors.newFixedThreadPool(numConsumers);
    
      final List<ConsumerLoop> consumers = new ArrayList<>();
      for (int i = 0; i < numConsumers; i++) {
        ConsumerLoop consumer = new ConsumerLoop(i, groupId, topics);
        consumers.add(consumer);
        executor.submit(consumer);
      }
    
      Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
          for (ConsumerLoop consumer : consumers) {
            consumer.shutdown();
          } 
          executor.shutdown();
          try {
            executor.awaitTermination(5000, TimeUnit.MILLISECONDS);
          } catch (InterruptedException e) {
            e.printStackTrace;
          }
        }
      });
    }

这个例子向一个Executor提交了3个Runnable Consumer。每一个Thread都给一个不同的Id，以便你可以发现是哪个Thread正在接收数据。当你停止处理时，shutdown hook就会被调用，这样就可以使用wakeup()停止3个Thread，并且等待它们shutdown。如果你运行它，你将会看来来自这些Thread的很多数据。

2: {partition=0, offset=928, value=2786}
    2: {partition=0, offset=929, value=2789}
    1: {partition=2, offset=297, value=891}
    2: {partition=0, offset=930, value=2792}
    1: {partition=2, offset=298, value=894}
    2: {partition=0, offset=931, value=2795}
    0: {partition=1, offset=278, value=835}
    2: {partition=0, offset=932, value=2798}
    0: {partition=1, offset=279, value=838}
    1: {partition=2, offset=299, value=897}
    1: {partition=2, offset=300, value=900}
    1: {partition=2, offset=301, value=903}
    1: {partition=2, offset=302, value=906}
    1: {partition=2, offset=303, value=909}
    1: {partition=2, offset=304, value=912}
    0: {partition=1, offset=280, value=841}
    2: {partition=0, offset=933, value=2801}

以上输出表明消费发生在三个Partition。每个Partition被分配给3个Thread中的一个。在每个Partition中，你可以看到offset如你所预料的那样增加。你可以用Ctrl+C或者IDE关闭程序。

# Consumer Liveness #

作为Consumer Group的一部分，每个Consumer被分配它已subscribe的Topic的Partitions的一个Partition子集。在这些Partition上有一个Group锁。如果这个锁被获得，该Group中的其他Consumer将不能再从锁住的Partition读数据。当你的Consumer是健康的时，这也正是你想要的。它是唯一的方法让你避免重复消费。但是，如果Consumer由于机器或者程序的失败而宕掉，你需要把锁释放掉，以便Partition可以分配给其他健康的Consumer。

Kafka的Group Coordination Protocol使用心跳机制（HeartBeat Mechanism）定位这个问题。在每次Rebalance之后，所有当前的Consumer都开始发送周期性心态给Group Coordinator。如果这个Coordinator一直接收心跳，它就假定这些Consumer是健康的。在每一个接收到的心跳中，该Coordinator都会开始（或重置）一个计时器。如果这个计时器过期，没有心跳接收，那么该Coordinator就会标记这个Consumer，给Group中的其他Consumer发信号，它们应该重新加入这个Group以便Partition重新分配。计时器的持续时间被称为Session Timeout，它在Client用 **session.timeout.ms**配置。

props.put("session.timeout.ms", "60000");

如果机器或者程序宕掉或者网络分区隔离了来自Coordinator的Consumer，Session Timeout确保锁将会被释放。但是，一般程序失败是一个小问题。因为Consumer一直向Coordinator发送心跳，这不代表程序是健康的。

Consumer的poll循环被设计成可以处理这个问题。当你调用poll()或者其他的阻塞API，所有的网络IO将在前台完成。Consumer不使用任何后台线程。这意味着当你调用poll()时，心跳只被发送给Coordinator。如果你的程序停止poll()（如，因为处理程序抛出异常，或者下游系统瘫痪），接着就没有心跳发送，Session Timeout过期，并且Group会被Rebalance。

唯一的问题是伪Rebalance可能因为Consumer处理message的时长大于Session Timeout被触发。因为，你需要设置Session Timeout足够长来处理message。它的默认值是30秒，但是设置几分钟也不是合理的。较大的Session Timeout的缺点是Coordinator检测真正的Consumer宕掉需要花费更长的时间。

# Delivery Semantics #

当Consumer第一次被创建时，根据通过配置auto.offset.reset的策略来初始化offset。一旦Consumer开始消费数据，它根据程序的需要周期性的提交offset。在以后每次Rebalance之后，对于Group中的Partition，其Position被设为最后一次提交的offset。如果在为已成功处理的message提交offset之前，Consumer宕掉，其他的Consumer将停止重复消费。你提交offset越频繁，在宕机后你看到的副本越少。

到目前为止，在例子中，我们都使用了自动提交的策略。当我们看到 **enable.auto.commit**被设置成true（默认值是true），Consumer根据配置的 **auto.commit.interval.ms**的Interval自动触发offset的提交。通过降低提交的Interval，你可以限制在宕机情况下必须做的重新处理的Consumer的数量。

为了使用自定义的提交API，你应该首先在Consumer的配置中通过设置 **enable.auto.commit**为false来禁用自动提交。

props.put("enable.auto.commit", "false");

提交的API是很容易被使用，但是最重要的一点是怎么集成在poll循环中。因此，下面的例子中包含拥有提交的完整的poll循环。操作手动提交的最简单的方法是使用同步方式提交APi。

try {
      while (running) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records)
          System.out.println(record.offset() + ": " + record.value());
    
        try {
          consumer.commitSync();
        } catch (CommitFailedException e) {
          // application specific failure handling
        }
      }
    } finally {
      consumer.close();
    }

例子使用没有参数的commitSync API提交最后一次调用poll()返回的offset。这个无限地阻塞调用直到要么提价成功，要么有一个不可恢复的失败提交。你需要关心的最主要的错误是当message的处理超过的Session Timeout的时长。当这种情况发生时，Coordinator将会把这个Consumer提出该Group，同时导致CommitFailException异常抛出。你的程序需要处理这个错误，通过回滚从最后一次成功提交的offset开始被消费过的message造成的所有改变。

你需要确定offset被提交只会发生在message被成功处理之后。如果在提交发送之前，Consumer就当掉了，这个message将不得不重新处理。如果提交策略保证了最后一次提交的offset不会在Current Position之前，那么你就是一个“at least once”的发送方式。

![Center 2][]

通过改变提交策略保证Current Position不会超过最后一次提交的offset（Last Committed Offset）。如上图所示，你就是“at most once”的发送方式。如果Consumer在它的Position赶上最后一次提交的offset之前就宕掉，那么所有message将会在这一阶段丢失，但是你可以确保没有message会被再次处理。为了实现这一策略，我们只需要改变提交和message处理的顺序。

try {
      while (running) {
      ConsumerRecords<String, String> records = consumer.poll(1000);
    
      try {
        consumer.commitSync();
        for (ConsumerRecord<String, String> record : records)
          System.out.println(record.offset() + ": " + record.value());
        } catch (CommitFailedException e) {
          // application specific failure handling
        }
      }
    } finally {
      consumer.close();
    }

注意使用自动提交，系统给你的是“at least once”策略，因为Consumer保证只为已经返回到你的应用程序的message提交offset。在最糟糕的情况下，你可能重新处理的message的数量上限，是你的程序在Commit Interval（通过 **auto.commit.interval.ms**配置）能够处理的message的数量。

然而，通过使用Commit API，你会有很多灵活的操作可以控制你将接受多少重复处理。最极端的情况是，你在每个message被处理之后提交offset，如下：

try {
      while (running) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
    
        try {
          for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.offset() + ": " + record.value());
            consumer.commitSync(Collections.singletonMap(record.partition(), new OffsetAndMetadata(record.offset() + 1)));
          }
        } catch (CommitFailedException e) {
          // application specific failure handling
        }
      }
    } finally {
      consumer.close();
    }

在这个例子中，我们在调用commitSync()方法中传入我们明确指定的想要提交的offset。这个提交的offset总是你的程序读到的下一条message的offset。当commitSync()方法没有参数时，Consumer提交被返回到你的程序的最后一次的offset（加上1）。但是，我们不能在这里使用，因为它需要Commit Position在我们实际处理的offset之前。

很明显，在每个message之后提交很可能对于很多场景不是一个好的想法，因为处理线程不得不等待每个提交请求从服务器返回而阻塞。这将降低消费能力。更合理的方法是在N个message之后提交，N个message可以协调很好性能。

这个例子中commitSync()的参数是一个出自Topic Partition的OffsetAndMetadata的实例的Map。Commit APi允许你在每次提交中添加一些额外的metadata。这样可以记录提交的时间，哪台主机发送的，或一切你的程序需要的信息。在这个例子中，我们没添加什么。

代替当message接收时提交，这里有个有一个更合理的策略：当你处理完来自每个Partition的message时，提交offset。ConsumerRecord集合提供了访问它包含的Partition集合，和每个Partition的message。下面的例子展示这种策略：

try {
      while (running) {
        ConsumerRecords<String, String> records = consumer.poll(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();
    }

到目前为止，这些例子都聚焦于同步方式提交的API，但是Consumer也可以利用异步方式的API——commitAsync。一般使用异步方式提交可以提高消费能力，因为你的程序可以在提交返回之前，就开始处理下一批message。需要权衡的是，你可能只能更晚地发现该提交失败。下面的例子给出基本的使用。

try {
      while (running) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records)
          System.out.println(record.offset() + ": " + record.value());
    
        consumer.commitAsync(new OffsetCommitCallback() {
          @Override
          public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, 
                                 Exception exception) {
            if (exception != null) {
              // application specific failure handling
            }
          }
        });
      }
    } finally {
      consumer.close();
    }

注意我们为commitAsync提供了回调，它当提交完成（无论成功或者失败）时被Consumer调用。如果你不需要这些，你也可以调用无参的commitAsync。

# Consumer Group检查 #

当一个Consumer Group是活跃的，你可以在命令行使用 consumer-groups.sh 脚本检查Partition的分配和消费的进度。它在Kafka根目录的bin文件夹中。

# bin/kafka-consumer-groups.sh --new-consumer --describe --group consumer-tutorial-group --bootstrap-server localhost:9092

其输出结果：

GROUP, TOPIC, PARTITION, CURRENT OFFSET, LOG END OFFSET, LAG, OWNER
    consumer-tutorial-group, consumer-tutorial, 0, 6667, 6667, 0, consumer-1_/127.0.0.1
    consumer-tutorial-group, consumer-tutorial, 1, 6667, 6667, 0, consumer-2_/127.0.0.1
    consumer-tutorial-group, consumer-tutorial, 2, 6666, 6666, 0, consumer-3_/127.0.0.1

它显示了该Consumer Group中的所有Partition分配，哪个Consumer拥有它，和最后一次提交的offset（即这里的“CURRENT OFFSET”）。一个Partition的滞后是The Log End Offset和The Last Committed Offset之差。管理员可以监视这个，以确保Consumer Group的消费跟得上Producer。

# 使用手动分配 #

正如本文开始提到的那样，新的Consumer为不需要Consumer Group的用户场景实现了“lower level”访问。这样的方便性是采用这种API最重要的原因之一。旧的“simple”Consumer也提供了这个，但是它要求你自己做很多错误处理。在新的Consumer中，你仅仅需要分配你想要读数据的Partition，并且开始poll。

下面的例子展示了使用partitionFor API分配一个Topic中的所有Partition。

List<TopicPartition> partitions = new ArrayList<>();
    for (PartitionInfo partition : consumer.partitionsFor(topic))
      partitions.add(new TopicPartition(topic, partition.partition()));
    consumer.assign(partitions);

类似于subscribe，调用assign()方法时必须传入你想要读数据的所有Partition的一个列表。一旦Partition被分配，poll循环就会向之前一样工作。

切记所有的offset提交都通过Group Coordinator，无论它是Simple Consumer还是Consumer Group。因此，如果你需要提交offset，你必须为 **group.id**设置一个合理的值，以免它和其他Consumer冲突。如果Simple Consumer用匹配活跃Consumer Group的一个Group Id提交offset，这个Coordinator将会拒绝提交（它将会导致CommitFailException）。如果另一个Simple Consumer实例共享同一个Group Id，将不会有任何错误发生。

[Center]: /images/20220721/f4717f0e99e84a009ecbc5df3a7d9249.png
[Center 1]: /images/20220721/5c77533f89394395a64792928856555c.png
[Center 2]: /images/20220721/e33cdc6eee61431abc564058c35c8a57.png