Kafka学习 — 3 kafka消费者客户端

二、消费者组和订阅主题

2.1 消费者组

消费者组：拥有相同group.id的消费者同属于一个消费者组；
kafka会将已订阅topic的消息发送到每个消费者组中。并通过平衡分区在消费者组中所有成员之间来达到平均。因此每个分区恰恰地分配1个消费者（一个消费者组中）。所有如果一个topic有4个分区，并且一个消费者分组有只有2个消费者。那么每个消费者将消费2个分区。
消费者组的成员是动态维护的：如果一个消费者故障。分配给它的分区将重新分配给同一个分组中其他的消费者。同样的，如果一个新的消费者加入到分组，将从现有消费者中移一个给它。这被称为重新平衡分组。
当分组重新分配自动发生时，可以通过ConsumerRebalanceListener关照消费者，这答应他们完成必要的应用程序级逻辑，例如状态清除，手动偏移提交等。
消费者也可以通过使用assign(Collection)手动分配指定分区，如果使用手动指定分配分区，那么动态分区分配和协调消费者组将失效。
2.2 发现消费者故障

1、订阅一组topic后，当调用poll(long）轮询时，消费者将自动加入到组中。只要持续的调用poll轮询，消费者将不绝保持可用，并继承从分配的分区中吸收消息。别的，消费者向服务器定时发送心跳。 如果消费者瓦解或无法在session.timeout.ms配置的时间内发送心跳，则消费者将被视为殒命，并且其分区将被重新分配。
2、另有一种可能，消费可能碰到“活锁”的情况，虽然它持续的发送心跳，但是没有处理。为了预防消费者在这种情况下不绝持有分区，我们使用max.poll.interval.ms活跃检测机制。 在此基础上，如果你调用的poll的频率大于最大间隔，则客户端将自动地离开组，以便其他消费者接受该分区。 发生这种情况时，你会看到offset提交失败（调用commitSync（）引发的CommitFailedException）。这是一种安全机制，保障只有活动成员可以或许提交offset。所以要留在组中，你必须持续调用poll轮询。
2.3消费者提供两个配置设置来控制poll循环：

2.3.1 max.poll.interval.ms：增大poll的间隔，可以为消费者提供更多的时间行止理返回的消息（调用poll(long)返回的消息，通常返回的消息都是一批）。缺点是此值越大将会延迟组重新平衡。
max.poll.records：此设置限制每次调用poll返回的消息数，如许可以更容易的猜测每次poll间隔要处理的最大值。通过调解此值，可以减少poll间隔，减少重新平衡分组的。
三、代码示例

3.1 手动提交偏移量

1. Properties props = new Properties();
2.      props.put("bootstrap.servers", "localhost:9092");
3.      props.put("group.id", "test");
4.      props.put("enable.auto.commit", "false");
5.      props.put("auto.commit.interval.ms", "1000");
6.      props.put("session.timeout.ms", "30000");
7.      props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
8.      props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
9.      KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
10.      consumer.subscribe(Arrays.asList("foo", "bar"));
11.      final int minBatchSize = 200;
12.      List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
13.      while (true) {
14.          ConsumerRecords<String, String> records = consumer.poll(100);
15.          for (ConsumerRecord<String, String> record : records) {
16.              buffer.add(record);
17.          }
18.          if (buffer.size() >= minBatchSize) {
19.              insertIntoDb(buffer);
20.              consumer.commitSync();
21.              buffer.clear();
22.          }
23.      }

复制代码

在这个示例中，当吸收的消息达到肯定数量后将它们批量插入到数据库中。如果我们设置offset自动提交（之前说的例子），默认认为消费已消费完成这批消息。但有可能在批处理记录之后，插入到数据库之前失败了，因为已经自动提交，如许就没法再继承获取之前的消息。
为了避免这种情况，我们应该再记录插入数据库之后再手动提交偏移量。如许可以正确控制消息是成功消费的。提出一个相反的可能性：在插入数据库之后，但是在提交之前，这个过程可能会失败（即使这可能只是几毫秒，这是一种可能性）。在这种情况下，进程将获取到已提交的偏移量，并会重复插入的最后一批数据。这种方式就是所谓的“至少一次”保证，在故障情况下，可以重复。
使用手动偏移控制的优点是可以直接控制消息何时被视为“已消费”。
3.2 风雅控制提交的偏移量

在某些情况下，你可以希望更风雅的控制，通过指定一个明确消息的偏移量为“已提交”。在下面，我们的例子中，我们处理完每个分区中的消息后，提交偏移量。

1. try {
2.          while(running) {
3.              ConsumerRecords<String, String> records = consumer.poll(Long.MAX\_VALUE);
4.              for (TopicPartition partition : records.partitions()) {
5.                  List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
6.                  for (ConsumerRecord<String, String> record : partitionRecords) {
7.                      System.out.println(record.offset() + ": " + record.value());
8.                  }
9.                  long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
10.                  consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
11.              }
12.          }
13.      } finally {
14.        consumer.close();
15.      }

复制代码

四、订阅指定的分区

在前面的例子中，我们订阅我们感兴趣的topic，让kafka提供给我们平分后的topic分区。但是，在有些情况下，你可能需要本身来控制分配指定分区，例如：
4.1 如果这个消费者进程与该分区生存了某种当地状态（如当地磁盘的键值存储），则它应该只能获取这个分区的消息。
4.2 如果消费者进程本身具有高可用性，运行失败后会自动重新启动（可能使用集群管理框架如YARN，Mesos，或者AWS设施，或作为一个流处理框架的一部分）。 在这种情况下，不需要Kafka检测故障，重新分配分区，因为消费者进程将在另一台呆板上重新启动。
要使用此模式，你只需调用assign（Collection）消费指定的分区即可：

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

复制代码

一旦手动分配分区，你可以在循环中调用poll（跟前面的例子一样）。消费者分组仍需要提交offset，只是如今分区的设置只能通过调用assign修改，因为手动分配分区后不会举行分组协调，因此消费者故障不会引发分区重新平衡。每一个消费者是独立工作的（即使和其他的消费者共享GroupId）。为了避免offset提交辩论，通常你需要确认每一个consumer实例的gorupId都是唯一的。
注意：手动分配分区（即，assgin）和动态分区分配的订阅topic模式（即，subcribe）不能混合使用。
五、offset的存储地方

消费者可以不使用kafka内置的offset仓库。可以选择本身来存储offset。
5.1 消费结果（数据）和offset存储在数据库中

如果消费的结果和offset存储在关系数据库中，需要将提交结果和offset在单个事务中。如许，事物成功，则offset存储和更新。如果offset没有存储，那么偏移量也不会被更新。
5.2 每个消费者都有本身的offset，所以要管理本身的偏移，你只需要做到以下几点：

5.2.1配置 enable.auto.commit=false；
5.2.2使用提供的 ConsumerRecord 来生存你的位置。
5.2.3在重启时用 seek(TopicPartition, long) 恢复消费者的位置。
六、消费者控制消费的位置

大多数情况下，消费者只是简朴的从头至尾的消费消息，周期性的提交位置（自动或手动）。kafka也支持消费者去手动的控制消费的位置，可以消费之前的消息也可以跳过近来的消息。
有几种情况，消费者手动控制消费的位置可能是有用的：
6.1、一种场景是对于时间敏感的消费者处理程序，对足够落伍的消息，直接跳过，从近来的消息开始消费。
6.2、另一个使用场景是当地状态存储体系（上一节说的）。在如许的体系中，消费者将要在启动时初始化它的位置（无论当地存储是否包含）。同样，如果当地状态已被粉碎（假设因为磁盘丢失），则可以通过重新消费所有数据并重新创建状态（假设kafka保存了足够的汗青）在新的呆板上重新创建。
6.3、 kafka使用seek(TopicPartition, long)指定新的消费位置。
用于查找kafka服务器保存的最早和最新的消息可用（seekToBeginning(Collection) 和 seekToEnd(Collection)）。
七、消费者控制流量

如果消费者分配了多个分区，并同时消费所有的分区，这些分区具有相同的优先级。在一些情况下，消费者需要首先消费一些指定的分区，当指定的分区有少量或者已经没有可消费的数据时，则开始消费其他分区。
例如流处理，当处理器从2个topic获取消息并把这两个topic的消息合并，当其中一个topic长时间落伍另一个，则停息消费，以便落伍的赶上来。
kafka支持动态控制流量，分别在future的poll(long)中使用pause(Collection) 来停息消费指定分配的分区，和 resume(Collection) 重新开始消费指定停息的分区。
八、多线程处理

Kafka消费者不是线程安全的。所有网络I/O都发生在举行调用应用程序的线程中。用户的责任是确保多线程访问正确同步的。非同步访问将导致ConcurrentModificationException。
此规则唯一的例外是wakeup()，它可以安全地从外部线程来停止活动操作。在这种情况下，将从操作的线程壅闭并抛出一个WakeupException。这可用于从其他线程来关闭消费者。 以下代码段表现了典范模式：

1. public class KafkaConsumerRunner implements Runnable {
2.      private final AtomicBoolean closed = new AtomicBoolean(false);
3.      private final KafkaConsumer consumer;
5.      public void run() {
6.          try {
7.              consumer.subscribe(Arrays.asList("topic"));
8.              while (!closed.get()) {
9.                  ConsumerRecords records = consumer.poll(10000);
10.                  // Handle new records
11.              }
12.          } catch (WakeupException e) {
13.              // Ignore exception if closing
14.              if (!closed.get()) throw e;
15.          } finally {
16.              consumer.close();
17.          }
18.      }
20.      // Shutdown hook which can be called from a separate thread
21.      public void shutdown() {
22.          closed.set(true);
23.          consumer.wakeup();
24.      }
25. }

复制代码

在单独的线程中，可以通过设置关闭标记和唤醒消费者来关闭消费者。

1.      closed.set(true);
2.      consumer.wakeup();

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。