课程表

Apache Kafka课程

工具箱
速查手册

Kafka 简单生产者示例

当前位置:免费教程 » 大数据/云 » Apache Kafka

让我们使用Java客户端创建一个用于发布和使用消息的应用程序。 Kafka生产者客户端包括以下API。

KafkaProducer API

让我们了解本节中最重要的一组Kafka生产者API。 KafkaProducer API的中心部分是 KafkaProducer 类。 KafkaProducer类提供了一个选项,用于将其构造函数中的Kafka代理连接到以下方法。

  • KafkaProducer类提供send方法以异步方式将消息发送到主题。 send()的签名如下

  1. producer.send(new ProducerRecord<byte[],byte[]>(topic,
  2. partition, key1, value1) , callback);
  • ProducerRecord - 生产者管理等待发送的记录的缓冲区。

  • 回调 - 当服务器确认记录时执行的用户提供的回调(null表示无回调)。

  • KafkaProducer类提供了一个flush方法,以确保所有先前发送的消息都已实际完成。 flush方法的语法如下 -

  1. public void flush()
  • KafkaProducer类提供了partitionFor方法,这有助于获取给定主题的分区元数据。 这可以用于自定义分区。 这种方法的签名如下 -

  1. public Map metrics()

它返回由生产者维护的内部度量的映射。

  • public void close() - KafkaProducer类提供关闭方法块,直到所有先前发送的请求完成。

生产者API

生产者API的中心部分是生产者类。 生产者类提供了一个选项,通过以下方法在其构造函数中连接Kafka代理。

生产者类

生产者类提供send方法以使用以下签名向单个或多个主题发送消息。


  1. public void send(KeyedMessaget<k,v> message)
  2. - sends the data to a single topic,par-titioned by key using either sync or async producer.
  3. public void send(List<KeyedMessage<k,v>>messages)
  4. - sends data to multiple topics.
  5. Properties prop = new Properties();
  6. prop.put(producer.type,"async")
  7. ProducerConfig config = new ProducerConfig(prop);

有两种类型的生产者 - 同步异步

相同的API配置也适用于同步生产者。 它们之间的区别是同步生成器直接发送消息,但在后台发送消息。 当您想要更高的吞吐量时,异步生产者是首选。 在以前的版本,如0.8,一个异步生产者没有回调send()注册错误处理程序。 这仅在当前版本0.9中可用。

public void close()

生产者类提供关闭方法以关闭与所有Kafka代理的生产者池连接。

配置设置

下表列出了Producer API的主要配置设置,以便更好地理解 -

S.No配置设置和说明
1

client.id

标识生产者应用程序

2

producer.type

同步或异步

3

acks

acks配置控制生产者请求下的标准是完全的。

4

重试

如果生产者请求失败,则使用特定值自动重试。

5

bootstrapping代理列表。

6

linger.ms

如果你想减少请求的数量,你可以将linger.ms设置为大于某个值的东西。

7

key.serializer

序列化器接口的键。

8

value.serializer

值。

9

batch.size

缓冲区大小。

10

buffer.memory

控制生产者可用于缓冲的存储器的总量。

ProducerRecord API

ProducerRecord是发送到Kafka cluster.ProducerRecord类构造函数的键/值对,用于使用以下签名创建具有分区,键和值对的记录。

  1. public ProducerRecord (string topic, int partition, k key, v value)
  • 主题 - 将附加到记录的用户定义的主题名称。

  • 分区 - 分区计数。

  • - 将包含在记录中的键。

  • 值 记录内容。
  1. public ProducerRecord (string topic, k key, v value)

ProducerRecord类构造函数用于创建带有键,值对和无分区的记录。

  • 主题 - 创建主题以分配记录。

  • - 记录的键。

  • - 记录内容。

  1. public ProducerRecord (string topic, v value)

ProducerRecord类创建一个没有分区和键的记录。

  • 主题 - 创建主题。

  • - 记录内容。

ProducerRecord类方法列在下表中 -

S.No类方法和描述
1

public string topic()

主题将附加到记录。

2

public K key()

将包括在记录中的键。 如果没有这样的键,null将在这里重新打开。

3

public V value()

记录内容。

4

partition()

记录的分区计数

SimpleProducer应用程序

在创建应用程序之前,首先启动ZooKeeper和Kafka代理,然后使用create topic命令在Kafka代理中创建自己的主题。 之后,创建一个名为 Sim-pleProducer.java 的java类,然后键入以下代码。

  1. //import util.properties packages
  2. import java.util.Properties;
  3.  
  4. //import simple producer packages
  5. import org.apache.kafka.clients.producer.Producer;
  6.  
  7. //import KafkaProducer packages
  8. import org.apache.kafka.clients.producer.KafkaProducer;
  9.  
  10. //import ProducerRecord packages
  11. import org.apache.kafka.clients.producer.ProducerRecord;
  12.  
  13. //Create java class named “SimpleProducer"
  14. public class SimpleProducer {
  15. public static void main(String[] args) throws Exception{
  16. // Check arguments length value
  17. if(args.length == 0){
  18. System.out.println("Enter topic name");
  19. return;
  20. }
  21. //Assign topicName to string variable
  22. String topicName = args[0].toString();
  23. // create instance for properties to access producer configs
  24. Properties props = new Properties();
  25. //Assign localhost id
  26. props.put("bootstrap.servers", localhost:9092");
  27. //Set acknowledgements for producer requests.
  28. props.put("acks", “all");
  29. //If the request fails, the producer can automatically retry,
  30. props.put("retries", 0);
  31. //Specify buffer size in config
  32. props.put("batch.size", 16384);
  33. //Reduce the no of requests less than 0
  34. props.put("linger.ms", 1);
  35. //The buffer.memory controls the total amount of memory available to the producer for buffering.
  36. props.put("buffer.memory", 33554432);
  37. props.put("key.serializer",
  38. "org.apache.kafka.common.serializa-tion.StringSerializer");
  39. props.put("value.serializer",
  40. "org.apache.kafka.common.serializa-tion.StringSerializer");
  41. Producer<String, String> producer = new KafkaProducer
  42. <String, String>(props);
  43. for(int i = 0; i < 10; i++)
  44. producer.send(new ProducerRecord<String, String>(topicName,
  45. Integer.toString(i), Integer.toString(i)));
  46. System.out.println(“Message sent successfully");
  47. producer.close();
  48. }
  49. }

编译 - 可以使用以下命令编译应用程序。

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

执行 - 可以使用以下命令执行应用程序。

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

输出

  1. Message sent successfully
  2. To check the above output open new terminal and type Consumer CLI command to receive messages.
  3. >> bin/kafka-console-consumer.sh --zookeeper localhost:2181 topic <topic-name> from-beginning
  4. 1
  5. 2
  6. 3
  7. 4
  8. 5
  9. 6
  10. 7
  11. 8
  12. 9
  13. 10

简单消费者示例

到目前为止,我们已经创建了一个发送消息到Kafka集群的生产者。 现在让我们创建一个消费者来消费Kafka集群的消息。 KafkaConsumer API用于消费来自Kafka集群的消息。 KafkaConsumer类的构造函数定义如下。

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

configs - 返回消费者配置的地图。

KafkaConsumer类具有下表中列出的以下重要方法。

S.No方法和说明
1

public java.util.Set< TopicPar- tition> assignment()

获取由用户当前分配的分区集。

2

public string subscription()

订阅给定的主题列表以获取动态签名的分区。

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 sub-scribe(java.util.regex.Pattern pattern,ConsumerRebalanceLis-tener listener)

参数模式以正则表达式的格式引用预订模式,而侦听器参数从预订模式获取通知。

7

public void as-sign(java.util.List< TopicPartion> partitions)

向客户手动分配分区列表。

8

poll()

使用预订/分配API之一获取指定的主题或分区的数据。 如果在轮询数据之前未预订主题,这将返回错误。

9

public void commitSync()

提交对主题和分区的所有子编制列表的最后一次poll()返回的提交偏移量。 相同的操作应用于commitAsyn()。

10

public void seek(TopicPartition partition,long offset)

获取消费者将在下一个poll()方法中使用的当前偏移值。

11

public void resume()

恢复暂停的分区。

12

public void wakeup()

唤醒消费者。

ConsumerRecord API

ConsumerRecord API用于从Kafka集群接收记录。 此API由主题名称,分区号(从中接收记录)和指向Kafka分区中的记录的偏移量组成。 ConsumerRecord类用于创建具有特定主题名称,分区计数和< key,value>的消费者记录。 对。 它有以下签名。

  1. public ConsumerRecord(string topic,int partition, long offset,K key, V value)
  • 主题 - 从Kafka集群接收的使用者记录的主题名称。

  • 分区 - 主题的分区。

  • - 记录的键,如果没有键存在null将被返回。

  • - 记录内容。

ConsumerRecords API

ConsumerRecords API充当ConsumerRecord的容器。 此API用于保存特定主题的每个分区的ConsumerRecord列表。 它的构造器定义如下。

  1. public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
  2. <Consumer-Record>K,V>>> records)
  • TopicPartition - 返回特定主题的分区地图。

  • 记录 - ConsumerRecord的返回列表。

ConsumerRecords类定义了以下方法。

S.No方法和描述
1

public int count()

所有主题的记录数。

2

public Set partitions()

在此记录集中具有数据的分区集(如果没有返回数据,则该集为空)。

3

public Iterator iterator()

迭代器使您可以循环访问集合,获取或重新移动元素。

4

public List records()

获取给定分区的记录列表。

配置设置

Consumer客户端API主配置设置的配置设置如下所示 -

S.No设置和说明
1

引导代理列表。

2

group.id

将单个消费者分配给组。

3

enable.auto.commit

如果值为true,则为偏移启用自动落实,否则不提交。

4

auto.commit.interval.ms

返回更新的消耗偏移量写入ZooKeeper的频率。

5

session.timeout.ms

表示Kafka在放弃和继续消费消息之前等待ZooKeeper响应请求(读取或写入)多少毫秒。

SimpleConsumer应用程序

生产者应用程序步骤在此保持不变。 首先,启动你的ZooKeeper和Kafka代理。 然后使用名为 SimpleCon-sumer.java 的Java类创建一个 SimpleConsumer 应用程序,并键入以下代码。

  1. import java.util.Properties;
  2. import java.util.Arrays;
  3. import org.apache.kafka.clients.consumer.KafkaConsumer;
  4. import org.apache.kafka.clients.consumer.ConsumerRecords;
  5. import org.apache.kafka.clients.consumer.ConsumerRecord;
  6.  
  7. public class SimpleConsumer {
  8. public static void main(String[] args) throws Exception {
  9. if(args.length == 0){
  10. System.out.println("Enter topic name");
  11. return;
  12. }
  13. //Kafka consumer configuration settings
  14. String topicName = args[0].toString();
  15. Properties props = new Properties();
  16. props.put("bootstrap.servers", "localhost:9092");
  17. props.put("group.id", "test");
  18. props.put("enable.auto.commit", "true");
  19. props.put("auto.commit.interval.ms", "1000");
  20. props.put("session.timeout.ms", "30000");
  21. props.put("key.deserializer",
  22. "org.apache.kafka.common.serializa-tion.StringDeserializer");
  23. props.put("value.deserializer",
  24. "org.apache.kafka.common.serializa-tion.StringDeserializer");
  25. KafkaConsumer<String, String> consumer = new KafkaConsumer
  26. <String, String>(props);
  27. //Kafka Consumer subscribes list of topics here.
  28. consumer.subscribe(Arrays.asList(topicName))
  29. //print the topic name
  30. System.out.println("Subscribed to topic " &plus; topicName);
  31. int i = 0;
  32. while (true) {
  33. ConsumerRecords<String, String> records = con-sumer.poll(100);
  34. for (ConsumerRecord<String, String> record : records)
  35. // print the offset,key and value for the consumer records.
  36. System.out.printf("offset = %d, key = %s, value = %s\n",
  37. record.offset(), record.key(), record.value());
  38. }
  39. }
  40. }

编译 - 可以使用以下命令编译应用程序。

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

执行 - 可以使用以下命令执行应用程序

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

输入 - 打开生成器CLI并向主题发送一些消息。 你可以把smple输入为\'Hello Consumer\'。

输出 - 以下是输出。

  1. Subscribed to topic Hello-Kafka
  2. offset = 3, key = null, value = Hello Consumer
转载本站内容时,请务必注明来自W3xue,违者必究。
 友情链接:直通硅谷  点职佳  北美留学生论坛

本站QQ群:前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

W3xue 的所有内容仅供测试,对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接(目前9元/月)请联系QQ:27243702 沸活量
皖ICP备17017327号-2 皖公网安备34020702000426号