Kafka 0.9 新版本consumer客户端使用介绍

电玩女神 2022-06-07 08:41 206阅读 0赞

翻译自:   
[https://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client/][https_www.confluent.io_blog_tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client]

kafka最初时开发时, 所带的producer和consumer client都是Scala所写. 我们逐渐发现这些API具有一些限制. high-level的api支持consumer groups和故障转移, 但是不支持许多复杂的使用场景, 同时还有一个simple consumer client提供全部的控制, 但是需要用户管理故障转移和负责错误处理. 为了支持一些旧client api很难或者无法做到的用例场景, 我们决定重新设计client, 提供一组能够长期支持的新api.

第一步是在0.8.1重写了Producer API, 0.9版本作为第二个步骤引入了新的Consumer API. 新的API基于kafka本身提供的一个新的group coordination 协议构建, 有以下的优点:

*  清晰统一的API: 新的consumer结合了之前的”simple”和”high-level” consumer客户端, 不仅提供group coordination还可以进行自己控制消费策略的lower level访问
 *  减少了依赖: 新的consumer由纯Java所写, 不需要依赖Scala运行时环境, 也不需要Zookeeper, 意味着你的项目所需要的包会减少一些
 *  更好的安全性: Kafka 0.9版本实现的 [security extensions][]在新的consumer中提供支持
 *  新的consumer还添加了一组用于管理容错的consumer group的协议. 之前这个功能由一个重量级的Java client(与Zookeeper进行交互)来实现. 这种复杂的逻辑使得完整功能的客户端难以由其他语言实现. 由于引进新的协议, 这变得远比以前容易. 实际上已经还在C客户端上转向使用这个协议.

虽然新consumer使用了重新设计的API和一个新的协调协议, 但是和之前旧版本API概念并不是完全不同的, 所以对旧consumer熟悉的用户理解新的API并不会有什么困难. 然而还是有一些关于group管理和线程模型等的很微妙的细节值得特别注意. 这篇教程的目的就是覆盖新consumer api的使用以及解释所有的这些细节.

值得注意的是: 在重写的过程中, 新的consumer在稳定性方面仍然被视为beta. 我们在0.9.0分支中以及修复了很多重要的bug, 如果你在使用0.9.0.0发行版时遇到了问题, 我们希望你能够对那个分支进行测试. 如果仍然有问题, 请反馈到 [mail lists][]或[JIRA][].

## Getting Started ##

在开始编码之前, 我们先回顾一下一些基本概念. 在Kafka中, 每个topic被分成一组称为partitions的logs. Producer向这些logs的末尾写入消息, consumer则自己按自己的节奏读取log. Kafka通过在一个consumer group中分配partitions来伸缩topic的消费, consumer group是一组有同样group id的consumers. 下图表示的是一个拥有3个分区的topic, 以及一个有两个成员的consumer group. 每一个分区都只会分配给组内的一个成员.   
![Distributing Kafka Partitions among consumer groups][]

旧的consumer依赖于Zookeeper来进行group管理, 新的消费者使用kafka内部的group coordination协议. 对于每个group, brokers中的一个被选为group coordinator. 这个coordinator负责管理group的状态, 它的主要工作是当新的成员加入, 或者原本的成员离开, 或者topic的元数据发生了改变时, 协调partition分配. 重新分配partition这个过程被称为rebalancing the group.

当group第一次初始化时, consumer通常从分区的最开始或者最末尾开始读取. 每个partition log中的消息都是按顺序读取. 随着consumer的读取, 它将会commit它所成功处理的message的offset. 例如下图中, consumer的位置位于offset 6, 上一次commit的offset为1.   
![consumer's position in the log][consumer_s position in the log]

当一个partition被重分配给组内的另一个consumer, 其最初位置会被设置成上一次commit的offset. 如果上例中的consumer突然奔溃了, 那么接管这个partition的另外一个组成员会从offset 1继续消费. 在这种情况下,它会重新处理上一次提交位置到消费者奔溃的位置6的消息.

这张图还有两个log中比较重要的位置. log end offset是最后一条写入log的message的offset, 而high watermark是最后一条message成功复制到所有的log replica的message offset. 从consumer的角度来看, 所知道的最主要的事情就是其最多能够读到high watermark的message. 这能够阻止consumer读取到未被复制到其他broker的, 可能会丢失的message.

## Configuration and Initialization 配置和初始化 ##

开始使用consumer之前, 需要将kafka-clients依赖加入到你的项目中.

<dependency>
     <groupId>org.apache.kafka</groupId>
     <artifactId>kafka-clients</artifactId>
     <version>0.9.0.0-cp1</version>
    </dependency>
     
      
       1
       2
       3
       4
       5

consumer使用一个Properties file来进行构建. 以下提供了使用consumer group所需要的最小配置.

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);
     
      
       1
       2
       3
       4
       5
       6

和之前的consumer, producer一样, 我们需要为consumer提供一个broker的初始化列表, 以用于发现集群中其他的broker. 这个配置并不需要提供所有的broker, client会从给定的配置中发现所有的当前存活的broker. 这里我们假设broker运行在本地, 同时consumer还需要知道如何反序列化message key和value. 最后, 为了加入consumer group, 我们需要配置一个group id. 随着教程的继续, 我们会介绍更多的配置.

## \* Topic订阅\* ##

为了开始消费, 你必须指定你的应用需要读取的topics, 在下面的例子中, 我们订阅了topic foo和bar.

consumer.subscribe(Arrays.asList(“foo”, “bar”));
     
      
       1

订阅之后, consumer会与组内其他consumer协调来获取其分区分配. 这都是在你开始消费数据时自动处理的. 后面我们会说明如何使用assign api来手动分配分区, 但是要注意的是, 不能同时混合使用自动和手动的分配.

subscribe方法并不是递增的: 你必须包含所有你想要消费的topics. 你可以在任何时刻改变你想消费的topic集, 当你调用subscribe时, 之前订阅的topics列表会被新的列表取代.

## 基本的Poll事件循环 ##

consumer需要能够并行获取数据, 在众多brokers中获取多个topic的许多分区消息. 为了实现这个目的, consumer API的设计成风格类似于unix中的poll或者select调用: 一旦topic注册了, 所有的coordination, rebalancing和data fetch都是由一个处于循环中的poll调用来驱动. 这样就提供了一个能在一个线程里处理所有的IO的简单有效的实现.

在你订阅一个topic之后, 你需要启动这个event loop以获得partition分配和开始获取数据. 听起来很复杂, 但是所有你需要做的就只有调用poll, 然后consumer客户端本身负责处理其他的工作. 每一次poll调用都会返回从所分配的partition获取的一组消息(也许是空的). 下面的例子展示了一个基本的poll循环, 打印获取的records的offset和value.

poll API根据当前的位置返回records, 当group第一次创建时, 消费开始的位置会被根据reset policy(一般设置成从每个分区的最早的offset或者最新的offset开始)来设置. 只要consumer开始提交offset, 那么之后的rebalance都会重置消费开始位置到最新的被提交的offset. 传递给poll的参数是consumer其在当前位置等待有record返回时需要被阻塞的时间. 一旦有record时, consumer会立即返回, 如果没有record, 它将会等待直到超时.

consumer被设计成只在自己的线程中运行, 在没有外部同步措施的情况下, 在多线程中使用时不安全的, 同时也不建议这样做. 在这个例子中, 我们使用了一个flag来使得当应用停止时能够从循环中跳出. 当这个flag被另一个线程设置成false时, pool返回时循环会跳出, 无论返回什么record, 处理过程都会结束.这个的例子使用了一个相对较少的超时时间, 以使得关闭consumer并不会有太大的延时.

你还可以设置一个较长的timeout, 并且使用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();
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11

我们将timeout改为了Long.MAX\_VALUE, 意味着consumer会一直阻塞直到有record返回. 和前面设置flag不同, 用于触发shutdown的线程可以调用consumer.wakeup()来中断一次poll, 使其抛出WakeupExection. 这个API是线程安全的. 注意如果当前没有活跃的poll, 那么异常会在下一次poll调用时抛出. 在这个例子中, 我们捕捉这个异常, 阻止其继续传播.

## 完整代码 ##

在接下来的例子中, 我们将所有的代码块放到一起来构建一个task, 初始化consumer, 订阅一个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();
      }
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
       43
       44

为了测试这个例子, 你需要一个运行0.9.0.0的kafka发行版的broker, 以及一个提供一些string数据进行消费的topic. 写入一些string数据的最简单的方法就是通过`kafka-verifiable-producer.sh`脚本. 你需要确保topic有超过一个partition. 例如对于只有一个运行在localhost的kafka broker和Zookeeper, 你可以在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 group, 所有的成员都订阅我们刚刚创建的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;
          }
        }
      });
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28

这个例子将三个Runnable consumer提交到一个executor. 每一个线程都有一个单独的id, 所以你可以看出来哪个线程正在接收数据. 当你准备结束进程时, shutdown hook会被调用, 它会调用wakeup来终止三个线程, 并且等待它们关闭. 以下是程序运行的数据:

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}
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17

## 消费者存活 ##

每个consumer都是被分配它所订阅的topics中所有partitions的一部分partitions. 就像是对这些partitions的group lock. 只要这个锁被持有, 组内其他的成员就无法从这些分区中读取数据. 当consumer还存活着, 这就是你想要的, 这是能够避免重复消费的方法(加锁). 但是如果消费者因为机器或者应用的故障挂掉了, 那么你希望锁能够释放, 并且分区能够分配给其他存活的consumer.

Kafka group coordination协议通过心跳机制来解决这个问题. 在每次重分配之后, 当前所有的成员会周期性发送心跳给group coordinator. 只要coordinator接收心跳, 它就会假设成员是存活的. 每次接收到心跳, coordinator都会开始(或重置)一个timer. 当timer过期时, 如果还没有心跳收到, coordinator会将这个成员标记为挂掉, 然后通知其他组内其他成员需要重新分配分区. timer的时间被称为session timeout, 由客户端的session.timeout.ms进行设置.

props.put(“session.timeout.ms”, “60000”);
     
      
       1

session timeout能够保证, 当机器或者应用奔溃, 或者网络分区导致consumer与coordinator分隔开来时, 锁会被释放. 然而应用故障判断却有点讨巧, 因为有可能consumer一直在发送心跳给coordinator, 但并意味着应用的状态是正常的.

consumer的poll api就是设计用于解决这个问题的(发送心跳), 当你调用poll或者其他阻塞API时所有的网络IO都是在前台处理的, consumer并不使用任何后台线程. 意味着发送给coordinator的心跳只会在poll调用时被发送( 译者注: 这个在0.10.1.0版本改成后台发送心跳), 如果应用停止poll( 无论是处理record的代码抛出了异常或者系统奔溃了),那么心跳将会停止发送, 那么session timeout就会过期, 组内的分区就会重新分配.

唯一一个你需要注意的问题就是: 如果你处理消息的时间比session timeout要长的话, 那么会使得coordinator认为consumer已经挂了, 导致partition重新分配. 你应该将session timeout设置的足够大, 以使得这种情况不会发生. 默认session timeout是30s, 但是将它设置成几分钟是不可行的, 这样就会导致coordinator需要更长的时间来检测真正的consumer奔溃.

## 消息传递语义 ##

当一个consumer group被建立时, 最开始消费的offset位置是由`auto.offset.reset`配置来控制的. 一旦consumer开始消费, 它会根据应用的需要自动提交offset. 每次rebalance, 消费开始的position都会被设置成分区最后提交的offset. 如果consumer在为已经成功处理的message提交offset之前奔溃了, 那么重新分配到这个分区的consumer会重复消费这些消息. 所以, 你提交消息的频率越频繁, 那么因为奔溃而造成的重复消费就会越少.

到现在,我们都假定开启了自动提交. 当我们将`enable.auto.commit`设置成true(默认是true), consumer会根据`auto.commit.interval.ms`的设置, 来周期性自动触发offset提交. 通过减少commit间隔, 你可以减少在奔溃之后消费者需要重新处理的message数量.

为了使用consumer的commit API, 你应该首先通过在consumer的配置中设置`enable.auto.commit`为false来禁止自动commit.

props.put(“enable.auto.commit”, “false”);
     
      
       1

commit API本身用起来不难, 但是重要的是你如何把它放入到代码中. 下面使用同步commit API是手动commit最简单的方法.

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();
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15

使用无参的commitSync方法会提交上次poll返回的最新的offset. 这个调用是阻塞式的, 直到commit成功或者因为不可恢复的错误而失败. 你需要担心的最大的问题就是消息处理时间可能比session timeout长. 当这种情况发生时, coordinator会把这个consumer踢出group, 那么consumer就会抛出CommitFailedException. 你的应用应该处理这个错误, 回滚自从上次成功提交offset以来的消费的message造成的改变.

一般你应该在消息被成功处理完后才去提交offset. 如果consumer在commit发送之前就奔溃了, 那么message将会被重新处理. 如果commit策略保证最后提交的offset不会超过当前的消费position, 那么你就会获得”at least once(至少一次)”的消息分发语义.   
![consumer-atmostonce-1.png][]

\*Figure 3: 提交的offset超前于当前消费位置\*  
通过改变commit策略来保证当前的消费位置绝对不超过最后一次committed offset, 如上图所示, 那么你会获得”at most once(至多一次)”的语义. 如果消费者在处理到最后一次committed offset之前就奔溃了, 那么所有在这个间隔之间的数据都会丢失, 但是你可以确保没有消息会被处理超过一遍. 为了实现这种策略, 我们只需要改变commit和消息处理的顺序.

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();
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15

使用自动提交为你提供”at least once”的语义, 因为consumer保证只commit返回给应用的message的offset. 你可能会重复处理的message就被限定在, 两次commit时间间隔(由auto.commit.interval.ms设置)之间你的应用处理的message.

通过使用commit API, 你对可以接受多少重复处理有更好的控制. 在最极端的情况下, 你可以在每次处理完消息都提交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();
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16

在这个例子中, 我们在commitSync方法中显式的传递了我们想要提交的offset. 提交的offset应该是你的应用将要读取的下一条message的offset. 当调用无参的commitSync方法时, 就会提交返回给应用的最后一条message的offset(再加上1), 但是我们不能这样做, 因为这样会使得提交的位置超过实际处理的进度.

明显每消费一条消息就commit offset并不合适大多数场景, 因为处理线程必须阻塞, 直到commit请求从服务器返回. 这样会大大减少吞吐量. 一个更加合理的方式就是每N条记录就提交一次, N可以进行调整以获得更好的性能.

commitSync方法的参数是一个键是topic partition, 值是offsetAndMetadata实例的map. commit API允许在每次commit时包含一些Metadata, 这些数据可以是记录commit的时间, 发送commit的host地址, 或者任何你的应用需要的信息. 在这个例子中, 我们让它为空.

相对每次接收到消息都提交offset, 更加合理的方法就是处理完一个分区的消息后再commit offset. ConsumerRecords Collection提供访问其中包含的partitions集合的方法, 以及每个partition的messages的方法. 下面的例子就展示了用法.

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();
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15

到目前为止, 所有的例子都是关注于同步commit API, 但是consumer同样也有一个异步的API, commitAsync, 使用异步的commit可以使得你的应用拥有更高的吞吐量, 因为你可以不用等待commit返回, 开始处理下一批message. 需要权衡的就是你可能随后发现, commit失败了. 以下的例子展示了基本用法:

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();
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19

注意到对于commitAsync我们提供了一个callback, 这个方法在consumer commit结束时被调用(无论commit成功与否), 如果你不需要callback, 那么你可以调用无参的commitAsync.

## 查看Consumer Group信息 ##

当一个consumer group处于活跃时, 你可以在命令行中使用位于Kafka发行包bin目录下的consumer-groups.sh脚本获取partition assignment和消费进度.

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

例子输出结果如下所示:

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
     
      
       1
       2
       3
       4

上面显示了consumer group中所有的partition assignment, 哪个消费者实例拥有哪个分区, 以及最新的committed offset(这里的current offset). lag表示log end offset与最新的committed offset的差值. 管理员可以使用这个命令来监控以保证consumer group跟上了producer的进度.

## 使用手动分区分配 ##

在教程前面提到的, 新的consumer为一些并不需要consumer group的应用场景实现了更底层的访问. 推荐使用这个API的理由就是它的易用性. 旧的”simple” consumer同样也提供了这个功能, 但是需要你自己进行一大堆的异常处理. 使用新的consumer API, 你只需要分配你想要读取数据的partition, 然后开始poll data就可以了.   
下面的例子展示了如何好使用partitionFor API来将一个topic的所有分区分配给一个consumer.

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

和subscribe方法类似, assign方法需要传递你想要读取的partition的列表. 一旦partition被分配了, poll的工作方式就和之前讲述的一致.

需要注意的是, 无论是使用一个simple consumer或者是consumer group, 所有的offset commit都会经过group coordinator. 所以如果你想要commit offsets, 你仍然需要设置`group.id`, 以防止与其他consumer发生冲突. 如果一个simple consumer试着使用和一个活跃的consumer group相同的group id来commit offset, 那么coordinator会拒绝这个commit(提交的结果就是抛出CommitFailedException). 有相同的group id的simple consumer却不会有错误.

## Conclusion ##

新的consumer为Kafka社区带来很多的好处, 包括一个清晰的API, 更好的安全性以及更少的依赖. 这篇教程介绍了基本的应用, 集中于poll的语义,以及使用commit API来控制消息传递语义. 虽然还有很多细节没有涉及到, 但是这对于开始使用kafka consumer来说已经足够了. 尽管新的consumer还在开发中, 但是我们鼓励你试一试. 如果有出现问题, 请在mailing list上告诉我们.

[https_www.confluent.io_blog_tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client]: https://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client/
[security extensions]: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=51809888
[mail lists]: https://kafka.apache.org/contact.html
[JIRA]: https://issues.apache.org/jira
[Distributing Kafka Partitions among consumer groups]: /images/20220607/8015da8dda80464da58d57bd747e05f5.png
[consumer_s position in the log]: /images/20220607/a725b8ecfc4b402fb9506040ab88cae9.png
[consumer-atmostonce-1.png]: /images/20220607/370e1b9902e846f78dc10aee1035cd83.png