Apache Kafka Consumer 消费者集

柔情只为你懂 2021-11-17 18:32 371阅读 0赞

## 1.目标 ##

在我们的上一篇文章中，我们讨论了[Kafka Producer][]。今天，我们将讨论Kafka Consumer。首先，我们将看到什么是Kafka Consumer和Kafka Consumer的例子。之后，我们将学习Kafka Consumer Group。此外，我们将看到Kafka Consumer的消费者记录API和配置设置。  
创建Kafka Producer后，将消息发送到[Apache Kafka][]集群。现在，我们正在创建一个Kafka Consumer来使用来自Kafka集群的消息。  
所以，让我们详细讨论Kafka Consumer。

[![Apache Kafka Consumer][]][Apache Kafka Consumer]

Apache Kafka Consumer | 卡夫卡消费者集团

## 2.什么是卡夫卡消费者？ ##

从[Kafka Topics][]读取数据的应用程序就是我们所说的Consumer。基本上，Kafka Consumer订阅了Kafka集群中的一个或多个主题，然后进一步提供来自Kafka主题的令牌或消息。 [ ][Kafka Topics]  
此外，使用Heartbeat，我们可以了解Consumer与[Kafka Cluster][]的连接性。但是，让我们定义Heartbeat。它设置在Consumer，让Zookeeper或Broker Coordinator知道Consumer是否仍然连接到Cluster。因此，如果心跳不存在，Kafka Consumer将不再连接到群集。在这种情况下，经纪协调员必须重新平衡负载。此外，Heartbeat是群集的开销。此外，通过考虑数据吞吐量和开销，我们可以配置心跳为消费者的时间间隔。

[![Apache Kafka Consumer][Apache Kafka Consumer 1]][Apache Kafka Consumer 1]

什么是Apache Kafka Consumer

此外，我们可以对消费者进行分组，而Kafka中的消费者群体中的消费者可以共享他们订阅的Kafka主题的分区。要理解，如果主题中有N个分区，Kafka Consumer Group中的N个消费者和该组已订阅主题，则每个消费者将从主题的分区中读取数据。因此，我们可以说，这只是一个消费者可以成群结队的提醒。  
[让我们用命令修改Apache Kafka Operations][Apache Kafka Operations]  
要具体来说，要连接到Kafka集群并使用数据流，Kafka的Consumer API会有所帮助。  
下面是显示Apache Kafka Consumer的图片：

[![Apache Kafka Consumer的工作][Apache Kafka Consumer 2]][Apache Kafka Consumer 2]

Apache Kafka Consumer的工作

要订阅一个或多个主题并处理在应用程序中生成的记录流，我们使用此Kafka Consumer API。换句话说，我们使用KafkaConsumer API来使用来自Kafka集群的消息。而且，下面看KafkaConsumer类的构造函数。

1.      public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)

*  **CONFIGS**

返回消费者配置图。  
KafkaConsumer类有以下重要方法：1。public   
java.util.Set <TopicPar-tition> assignment（）  
获取当前由使用者分配的分区集。  
2. public string subscription（）  
为了订阅给定的主题列表，获取动态分配的分区。  
[探索Kafka性能调优 - Kafka优化的方法][Kafka_ - Kafka]  
3. public void sub-scribe（java.util.List <java.lang.String> topics，ConsumerRe-balanceListener listener）  
此外，订阅给定动态分配的主题列表分区。  
4。 public void unsubscribe（）  
现在，取消订阅给定分区列表中的主题。  
5. public void sub-scribe（java.util.List <java.lang.String> topics）  
为了订阅给定的主题列表以获取动态分配的分区。如果给定的主题列表为空，则将其视为与unsubscribe（）相同。  
6.  public void subscribe（java.util.regex.Pattern pattern，ConsumerRebalanceLis-tener listener）  
这里，参数模式引用正则表达式格式的订阅模式，并且listener参数从订阅模式获取通知。  
7. public void as-sign（java.util.List <TopicParti-tion> partitions）  
手动为客户分配分区列表。  
8.  民意调查（）  
获取使用其中一个subscribe / assign API指定的主题或分区的数据。如果在轮询数据之前未订阅主题，则会返回错误。  
9. public void commitSync（）  
为了提交所有订阅的主题和分区列表的最后一个poll（）中返回的偏移量。对commitAsyn（）应用相同的操作。  
10.  public void seek（TopicPartition partition，long offset）  
获取消费者将在下一个poll（）方法中使用的当前偏移值。  
[阅读Kafka的优点和缺点][Kafka]  
11. public void resume（）  
为了恢复暂停的分区。  
12. public void wakeup（）  
唤醒消费者。

## 3. ConsumerRecord API ##

基本上，要从Kafka集群接收记录，我们使用ConsumerRecord API。它包括一个主题名称，分区号，从中接收记录的偏移量也指向Kafka分区中的记录。此外，要创建具有特定主题名称，分区计数和<key，value>对的使用者记录，我们使用consumerRecord类。它的签名是：

1.  public ConsumerRecord （字符串主题，int分区，长偏移量，K键，V值）
2.      public ConsumerRecord(string topic,int partition, long offset,K key, V value)

*  话题

从Kafka群集收到的消费者记录的主题名称。

*  划分

主题的分区。

*  键

记录的密钥，如果没有密钥存在，则返回null。

*  值

记录内容。  
[学习Apache Kafka Streams | 流处理拓扑][Apache Kafka Streams]

## 4. ConsumerRecords API ##

基本上，它是ConsumerRecord的容器。要保留特定主题的每个分区的ConsumerRecord列表，我们使用此API。它的构造函数是：

1.      public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
        <Consumer-Record>K,V>>> records)

*  TopicPartition

返回特定主题的分区映射。

*  记录

返回ConsumerRecord列表。  
这些是ConsumerRecords类的以下方法：  
1。public  int count（）  
所有主题的记录数。  
2.  public Set partitions（）  
具有此记录集中数据的分区集（如果未返回任何数据，则该集为空）。  
3.  public Iterator iterator（）  
通常，迭代器使您可以遍历集合，获取或删除元素。  
4.  public list records（）  
基本上，获取给定分区的记录列表。

## 5. ConsumerRecord API与ConsumerRecords API ##

一个。ConsumerRecord API  
ConsumerRecord API是从Kafka接收的键/值对。它包含主题名称和分区号，从中接收记录以及指向Kafka分区中记录的偏移量。  
湾 ConsumerRecords API  
然而，ConsumerRecords API是一个容器，它为特定主题的每个分区保存ConsumerRecord列表。基本上，Consumer.poll（long）操作返回的每个主题分区都有一个ConsumerRecord列表。  
[Apache Kafka工作流程| Kafka Pub-Sub Messaging][Apache Kafka_Kafka Pub-Sub Messaging]

## 6.配置设置 ##

在这里，我们列出了Consumer客户端API的配置设置 -   
1.  bootstrap.servers  
它引导了代理列表。  
2.  group.id  
将个人消费者分配给一个组。  
3.  enable.auto.commit  
基本上，如果值为true，则启用偏移的自动提交，否则不提交。  
4.  auto.commit.interval.ms  
基本上，它返回更新的消耗偏移量写入[ZooKeeper的][ZooKeeper]频率。  
5.  session.timeout.ms  
它表示Kafka在放弃并继续使用消息之前等待ZooKeeper响应请求（读或写）的毫秒数。

## 7. SimpleConsumer应用程序 ##

确保生产者应用程序步骤在此处保持不变。这里也开始你的ZooKeeper和Kafka经纪人。此外，使用名为SimpleCon-sumer.java 的[java类][java]创建SimpleConsumer应用程序。然后键入以下代码：  
[阅读Apache Kafka职业范围与薪资趋势。][Apache Kafka 1]

import java.util.Properties;
    import java.util.Arrays;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    public class SimpleConsumer {
      public static void main(String[] args) throws Exception {
         if(args.length == 0){
            System.out.println("Enter topic name");
            return;
         }
         //Kafka consumer configuration settings
         String topicName = args[0].toString();
         Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("group.id", "test");
         props.put("enable.auto.commit", "true");
         props.put("auto.commit.interval.ms", "1000");
         props.put("session.timeout.ms", "30000");
         props.put("key.deserializer",
            "org.apache.kafka.common.serializa-tion.StringDeserializer");
         props.put("value.deserializer",
            "org.apache.kafka.common.serializa-tion.StringDeserializer");
         KafkaConsumer<String, String> consumer = new KafkaConsumer
            <String, String>(props);
         //Kafka Consumer subscribes list of topics here.
         consumer.subscribe(Arrays.asList(topicName))
         //print the topic name
         System.out.println("Subscribed to topic " + topicName);
         int i = 0;
         while (true) {
            ConsumerRecords<String, String> records = con-sumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
            // print the offset,key and value for the consumer records.
            System.out.printf("offset = %d, key = %s, value = %s\n",
               record.offset(), record.key(), record.value());
         }
      }
    }

###  A。汇编 ###

通过使用以下命令，我们可以编译应用程序。

1.    
    
    
        javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

###  B.执行 ###

而且，使用以下命令我们可以执行应用程序。

1.      java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleConsumer <topic-name>

###  C.输入 ###

此外，打开生产者CLI并向主题发送一些消息。我们可以将简单输入作为'Hello Consumer'。

### d. 产量 ###

该输出是

1.  订阅主题Hello-Kafka
2.  offset = 3 ，key = null，value = Hello Consumer
3.      Subscribed to topic Hello-Kafka
        offset = 3, key = null, value = Hello Consumer

## 8.卡夫卡消费者集团 ##

基本上，Kafka中的Consumer组是来自Kafka主题的多线程或多机器消费。

[![卡夫卡消费者 - 卡夫卡消费者集团][-]][-]

卡夫卡消费者 - 卡夫卡消费者集团

*  通过使用相同的group.id，消费者可以加入一个组。
 *  组的最大并行度是组中的消费者数量←分区数。
 *  此外，Kafka将主题的分区分配给组中的消费者。因此，每个分区仅由该组中的一个消费者使用。
 *  此外，Kafka保证消息只能由组中的单个消费者读取。
 *  消费者可以按照日志中存储的顺序查看消息。

[看看Storm Kafka与配置和代码][Storm Kafka]  
的[集成][Storm Kafka] a。重新平衡消费者  
基本上，增加更多流程/线程将导致Kafka重新平衡。基本上，如果任何消费者或代理以某种方式无法向ZooKeeper发送心跳，则可以通过Kafka群集重新配置它。此外，在此重新平衡期间，Kafka将可用分区分配给可用线程，可能将分区移动到另一个进程。

import java.util.Properties;
    import java.util.Arrays;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    public class ConsumerGroup {
      public static void main(String[] args) throws Exception {
         if(args.length < 2){
            System.out.println("Usage: consumer <topic> <groupname>");
            return;
         }
         String topic = args[0].toString();
         String group = args[1].toString();
         Properties props = new Properties();
         props.put("bootstrap.servers", "localhost:9092");
         props.put("group.id", group);
         props.put("enable.auto.commit", "true");
         props.put("auto.commit.interval.ms", "1000");
         props.put("session.timeout.ms", "30000");
         props.put("key.deserializer",
            "org.apache.kafka.common.serializa-tion.StringDeserializer");
         props.put("value.deserializer",
            "org.apache.kafka.common.serializa-tion.StringDeserializer");
         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
         consumer.subscribe(Arrays.asList(topic));
         System.out.println("Subscribed to topic " + topic);
         int i = 0;
         while (true) {
            ConsumerRecords<String, String> records = con-sumer.poll(100);
               for (ConsumerRecord<String, String> record : records)
                  System.out.printf("offset = %d, key = %s, value = %s\n",
                  record.offset(), record.key(), record.value());
         }
      }
    }
    ii. Compilation
    javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*" ConsumerGroup.java
    iii. Execution
    >>java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*":.
    ConsumerGroup <topic-name> my-group
    >>java -cp "/home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*":.
    ConsumerGroup <topic-name> my-group

因此，我们可以看到我们创建了名称的样本组，my-group和两个消费者。  
湾 输入  
现在，在打开生产者CLI后，发送一些消息，如 -

Test consumer group 01
    Test consumer group 02

1.  测试消费者组01
2.  测试消费者组02

C。第一个过程的输出  
[学习Apache Kafka用例| 卡夫卡应用程序][Apache Kafka 2]

Subscribed to topic Hello-kafka
    offset = 3, key = null, value = Test consumer group 01

1.  订阅主题Hello-kafka
2.  offset = 3 ，key = null，value =测试消费者组01

d。此外，第二个过程的输出

Subscribed to topic Hello-kafka
    offset = 3, key = null, value = Test consumer group 02

1.  订阅主题Hello-kafka
2.  offset = 3 ，key = null，value =测试消费者组02

所以，这完全是关于Kafka的Apache Kafka消费者和消费者群体的例子。希望你喜欢我们的解释。

## 9.结论：卡夫卡消费者 ##

因此，我们通过使用Java客户端演示详细了解了Kafka Consumer和ConsumerGroup。此外，通过这个，我们了解了如何使用Java客户端发送和接收消息。此外，我们讨论了Kafka Consumer记录API和Consumer Records API以及两者的比较。此外，我们还学习了Kafka Consumer客户端API的配置设置。但是，如果有任何疑问，请随时在评论部分询问。  
另请参阅 -    
[Kafka Broker ][Kafka Broker]  
[供参考][Link 1]

转载于:https://www.cnblogs.com/a00ium/p/10850141.html

[Kafka Producer]: https://data-flair.training/blogs/kafka-producer/
[Apache Kafka]: https://data-flair.training/blogs/apache-kafka-tutorial/
[Apache Kafka Consumer]: /images/20211116/235f49a05bf64d7aac22d3e6743cfae4.png
[Kafka Topics]: https://data-flair.training/blogs/kafka-topic-architecture/
[Kafka Cluster]: https://data-flair.training/blogs/kafka-cluster/
[Apache Kafka Consumer 1]: /images/20211116/0775c70e1c8e4269b6781c75b53c3ea6.png
[Apache Kafka Operations]: https://data-flair.training/blogs/kafka-operations/
[Apache Kafka Consumer 2]: /images/20211116/9674b5eb729043999d3a5f54d5d58e06.png
[Kafka_ - Kafka]: https://data-flair.training/blogs/kafka-performance-tuning/
[Kafka]: https://data-flair.training/blogs/advantages-and-disadvantages-of-kafka/
[Apache Kafka Streams]: https://data-flair.training/blogs/kafka-streams/
[Apache Kafka_Kafka Pub-Sub Messaging]: https://data-flair.training/blogs/kafka-workflow/
[ZooKeeper]: https://data-flair.training/blogs/zookeeper-tutorial/
[java]: https://data-flair.training/blogs/class-and-object-in-java/
[Apache Kafka 1]: https://data-flair.training/blogs/kafka-career/
[-]: /images/20211116/00028be228dc41a08c686d57837f8d9d.png
[Storm Kafka]: https://data-flair.training/blogs/storm-kafka-integration/
[Apache Kafka 2]: https://data-flair.training/blogs/kafka-use-cases-and-applications/
[Kafka Broker]: https://data-flair.training/blogs/kafka-broker/
[Link 1]: https://kafka.apache.org/0100/.../index.html?.../kafka/clients/consumer/KafkaConsume...