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

kafkaproducer api

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

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

producer.send(new producerrecord<byte[],byte[]>(topic, 
partition, key1, value1) , callback);

• producerrecord - 生产者管理等待发送的记录的缓冲区。

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

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

public void flush()

• kafkaproducer类提供了partitionfor方法，这有助于获取给定主题的分区元数据。 这可以用于自定义分区。 这种方法的签名如下 -

public map metrics()

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

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

生产者api

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

生产者类

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

public void send(keyedmessaget<k,v> message) 
- sends the data to a single topic,par-titioned by key using either sync or async producer.
public void send(list<keyedmessage<k,v>>messages)
- sends data to multiple topics.
properties prop = new properties();
prop.put(producer.type,"async")
producerconfig config = new producerconfig(prop);

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

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

public void close()

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

配置设置

下表列出了producer api的主要配置设置，以便更好地理解 -

producerrecord api

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

public producerrecord (string topic, int partition, k key, v value)

• 主题 - 将附加到记录的用户定义的主题名称。

• 分区 - 分区计数。

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

• 值 - 记录内容。

public producerrecord (string topic, k key, v value)

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

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

• 键 - 记录的键。

• 值 - 记录内容。

public producerrecord (string topic, v value)

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

• 主题 - 创建主题。

• 值 - 记录内容。

producerrecord类方法列在下表中 -

simpleproducer应用程序

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

//import util.properties packages
import java.util.properties;

//import simple producer packages
import org.apache.kafka.clients.producer.producer;

//import kafkaproducer packages
import org.apache.kafka.clients.producer.kafkaproducer;

//import producerrecord packages
import org.apache.kafka.clients.producer.producerrecord;

//create java class named “simpleproducer"
public class simpleproducer {
   
   public static void main(string[] args) throws exception{
      
      // check arguments length value
      if(args.length == 0){
         system.out.println("enter topic name");
         return;
      }
      
      //assign topicname to string variable
      string topicname = args[0].tostring();
      
      // create instance for properties to access producer configs   
      properties props = new properties();
      
      //assign localhost id
      props.put("bootstrap.servers", “localhost:9092");
      
      //set acknowledgements for producer requests.      
      props.put("acks", “all");
      
      //if the request fails, the producer can automatically retry,
      props.put("retries", 0);
      
      //specify buffer size in config
      props.put("batch.size", 16384);
      
      //reduce the no of requests less than 0   
      props.put("linger.ms", 1);
      
      //the buffer.memory controls the total amount of memory available to the producer for buffering.   
      props.put("buffer.memory", 33554432);
      
      props.put("key.serializer", 
         "org.apache.kafka.common.serializa-tion.stringserializer");
         
      props.put("value.serializer", 
         "org.apache.kafka.common.serializa-tion.stringserializer");
      
      producer<string, string> producer = new kafkaproducer
         <string, string>(props);
            
      for(int i = 0; i < 10; i++)
         producer.send(new producerrecord<string, string>(topicname, 
            integer.tostring(i), integer.tostring(i)));
               system.out.println(“message sent successfully");
               producer.close();
   }
}

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

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

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

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

输出

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

简单消费者示例

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

public kafkaconsumer(java.util.map<java.lang.string,java.lang.object> configs)

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

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

consumerrecord api

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

public consumerrecord(string topic,int partition, long offset,k key, v value)

• 主题 - 从kafka集群接收的使用者记录的主题名称。

• 分区 - 主题的分区。

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

• 值 - 记录内容。

consumerrecords api

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

public consumerrecords(java.util.map<topicpartition,java.util.list
<consumer-record>k,v>>> records)

• topicpartition - 返回特定主题的分区地图。

• 记录 - consumerrecord的返回列表。

consumerrecords类定义了以下方法。

配置设置

consumer客户端api主配置设置的配置设置如下所示 -

simpleconsumer应用程序

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

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 " &plus; 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());
      }
   }
}

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

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

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

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

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

输出 - 以下是输出。

subscribed to topic hello-kafka
offset = 3, key = null, value = hello consumer