配置使用者属性

对于控制其行为方面的使用者，存在许多配置设置。 以下设置是一些最重要的设置。

提供了许多其他配置设置，但请先阅读 Apache Kafka 文档，然后再开始对其进行试验。

使用者组

_消费者群组_是指合作消费一个或多个主题信息的消费者群组。 组中的使用者全部使用相同的 group.id 配置值。 如果您需要多个使用者来处理工作负载，可以在同一使用者组中运行多个使用者。 即使只需要一个消费者，通常也要为 group.id 指定一个值。

每个消费者群组都有一台服务器，称为_协调器_，负责为群组中的消费者分配分区。 这项责任会分布在集群中的各个服务器上，以均衡负载。 每次进行组重新均衡时，为使用者分配的分区都会更改。

使用者加入使用者组时，会发现该组的协调者。 然后，消费者告诉协调器它想加入组，协调器开始重新平衡整个组的分区，将新成员纳入其中。

来自单个分区的消息仅由每个组中的一个使用者进行处理。 这种机制可确保每个分区上的报文按顺序处理。 请参阅下图以获取一个示例，其中主题包含三个分区，而使用该主题的使用者组包含两个使用者。 为组中的一个使用者分配了两个分区，为另一个使用者分配了一个分区。

当用户组发生以下变化之一时，该组会通过向组成员分配分区来重新平衡，以适应变化：

• 一名消费者加入小组。
• 消费者离开小组。
• 消费者被协调员视为不再存活。
• 在现有主题中添加新分区。

如果你有一个重新平衡的消费者群组，请注意，任何离开群组的消费者在重新加入群组之前，其提交都会被拒绝。 在这种情况下，使用者需要重新加入该组，在该组中为其分配的分区可能会与先前使用的不同。

使用者活动性

Kafka 会自动检测失败的使用者，以便可以将分区重新分配给有效的使用者。 它使用两种机制: 轮询和脉动信号。

如果从 Consumer.poll(...) 返回的消息批处理较大或处理耗时，那么再次调用 poll() 之前的延迟可能很重要或不可预测。 在某些情况下，有必要配置较长的最大轮询时间间隔，这样就不会因为消息处理需要一段时间而将使用者从组中除去。 如果此机制是唯一可用的机制，那么检测失败的使用者所需的时间也很长。

为了更轻松地处理使用者活动性问题，Kafka 0.10.1 中添加了后台脉动信号发送功能。 组协调者期望组成员定期向其发送脉动信号，以指明自己仍然处于活动状态。 消费者中运行一个后台心跳线程，定期向协调者发送心跳。 如果协调器在_会话超时时间_内没有收到群成员的心跳，协调器就会将该成员从群中删除，并开始重新平衡群。 会话超时可以比最大轮询间隔短得多，这样，即使报文处理需要很长时间，检测到消费者失败所需的时间也可以很短。

您可以使用 max.poll.interval.ms 属性配置最大轮询间隔，使用 session.timeout.ms 属性配置会话超时。 除非处理一批邮件需要 5 分钟以上，否则不需要使用这些设置。

管理偏移量

对于每个消费者组，Kafka会为每个被消费的分区维护已提交偏移量。 使用者处理消息后，不会从分区中除去该消息。 相反，它只是通过一个称为提交偏移量的过程来更新当前偏移量。

Event Streams 会将已落实偏移量信息保留 7 天。

控制消息使用速度

如果由于信息泛滥导致信息处理出现问题，可以设置消费者选项来控制信息消耗的速度。 使用 fetch.max.bytes 和 max.poll.records 可控制调用一次 poll() 可以返回多少数据。

处理使用者重新均衡

当消费者添加到群组或从群组中删除时，群组会重新平衡，消费者无法消费信息。 这将导致消费者群体中的所有消费者在短时间内无法使用。

如果收到“分区已撤销”回调通知，请使用ConsumerRebalanceListener手动提交偏移量（如果不使用自动提交），并暂停进一步处理，直到收到“分区已分配”回调的成功重新平衡通知。

代码片段

这些代码片段都是高层次的，用于说明相关概念。 有关完整示例，请参阅 GitHub中的 Event Streams 样本。

要将使用者连接到 Event Streams，您需要创建服务凭证。 有关如何获取这些凭证的信息，请参阅 连接到 Event Streams。

在使用者代码中，首先需要构建配置属性集。 Event Streams的所有连接都使用 TLS 和用户密码验证，因此至少需要这些属性。 将 BOOTSTRAP_ENDPOINTS、USER 和 PASSWORD 替换为您自己服务证书中的内容：

Properties props = new Properties();
 props.put("bootstrap.servers", BOOTSTRAP_ENDPOINTS);
 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USER\" password=\"PASSWORD\";");
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "PLAIN");
 props.put("ssl.protocol", "TLSv1.2");
 props.put("ssl.enabled.protocols", "TLSv1.2");
 props.put("ssl.endpoint.identification.algorithm", "HTTPS");

要使用消息，还需要为键和值指定反序列化器，如以下示例中所示。

 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

这些反序列化器必须与生产者使用的序列化器匹配。

然后，通过 KafkaConsumer 来使用消息，其中每条消息由一个 ConsumerRecord 表示。 使用消息最常用的方法是通过设置组标识，将使用者放入使用者组中，然后调用 subscribe() 来获取主题列表。 消费者会被分配到一些分区进行消费，但如果组中的消费者多于主题中的分区，消费者可能不会被分配到任何分区。 接下来，在一个循环中调用 poll()，接收一批要处理的消息，其中每条消息由一个 ConsumerRecord 表示。

props.put("group.id", "G1");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));  // T1 is the topic name
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());
}

这个消费者循环将永远运行，但可以通过调用 Consumer.wakeup() 从另一个线程中断，从而实现整洁的关闭。

要手动落实偏移量，首先必须将 enable.auto.commit 配置设置为 false。 然后，使用 Consumer.commmitSync() 或 Consumer.commitAsync() 定期更新使用者的已落实偏移量。 为了简单起见，此示例处理每个分区的记录并分别落实最后一个偏移量。

props.put("group.id", "G1");
props.put("enable.auto.commit", "false");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> partRecords = records.records(tp);
      long lastOffset = 0;
      for (ConsumerRecord<String, String> record : partRecords) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        lastOffset = record.offset();
      }
      // having processed all the records in the above loop, we commit the partition's offset to 1 more than the last offset
      consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastOffset + 1)));
    }
  }
}
finally {
  consumer.close();
}

异常处理

任何使用 Kafka 客户机的稳健应用程序都需要处理某些预期情境的异常。 在某些情况下，异常不会被直接抛出，因为有些方法是异步的，会通过 Future 或回调来传递结果。 请查看 GitHub 中显示完整示例的示例代码。

处理代码中的以下异常列表: