在这一节中将创建一个使用java客户端发布和使用消息的应用程序。 kafka生产者客户端由以下api组成。
下面来了解kafka生产者api。 kafkaproducer api的核心部分是kafkaproducer类。 kafkaproducer类提供了一个选项，用于将kafka代理的构造函数与以下方法连接起来。
kafkaproducer类提供send()方法来异步发送消息到主题。 send()的签名如下 - producer.send(new producerrecord<byte[],byte[]>(topic, partition, key1, value1) , callback);
producerrecord - 生产者管理等待发送的记录缓冲区。callback - 服务器确认记录时执行的用户提供的回调函数(null表示无回调)。kafkaproducer类提供了一个flush方法来确保所有先前发送的消息已经实际完成。 flush方法的语法如下 -public void flush()
kafkaproducer类提供了partitionfor方法，该方法有助于获取给定主题的分区元数据。 这可以用于自定义分区。 这种方法的签名如下 -public map metrics()
public void close() - kafkaproducer类提供close方法，阻塞直到完成所有先前发送的请求。producer api的核心部分是producer类。 producer类提供了一个选项，通过以下方法在其构造函数中连接kafka代理。
producer类
producer类提供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配置也适用于sync生产者。 它们之间的区别是同步生产者直接发送消息，但在后台发送消息。 当想要更高的吞吐量时，首选异步生产者。 在0.8之前的版本中，异步生产者没有回调send()来注册错误处理程序。异步生产者仅在当前版本0.9中可用。
public void close()
producer类提供close()方法来关闭所有kafka 经纪人的生产者池连接。
producer api的主要配置设置在下表中列出以便更好地理解 -
| 编号 | 配置设置 | 描述 |
|---|---|---|
| 1 | client.id |
确定生产者应用 |
| 2 | producer.type |
是同步还是异步? |
| 3 | acks |
acks配置控制生产者请求下的标准被认为是完整的。 |
| 4 | retries |
如果生产者请求失败，则自动重试具有特定的值。 |
| 5 | bootstrap.servers |
引导经纪人(brokers)列表。 |
| 6 | linger.ms |
如果想减少请求的数量，可以将linger.ms设置为大于某个值的值。 |
| 7 | key.serializer |
序列化器接口的键。 |
| 8 | value.serializer |
序列化器接口的值。 |
| 9 | batch.size |
缓冲区大小。 |
| 10 | buffer.memory |
控制生产者可用于缓冲的总内存量。 |
producerrecord api
producerrecord是发送给kafka的cluster.producerrecord类的键/值对，用于使用以下签名创建包含分区，键和值对的记录。
public producerrecord (string topic, int partition, k key, v value)
topic − 用户定义的主题名称将要追加到记录中。partition − 分区计数key − 记录中将包含的关键字。value − 记录内容public producerrecord (string topic, k key, v value)
producerrecord类构造函数用于创建一个记录，其中包含键值对，但没有分区。
topic − 创建一个分配记录的主题。partition − 分区计数key − 记录中将包含的关键字。value − 记录内容public producerrecord (string topic, v value)
producerrecord类创建没有分区和键的记录。
topic − 创建一个分配记录的主题。value − 记录内容下表中列出了producerrecord类的方法 -
| 编号 | 方法 | 描述 |
|---|---|---|
| 1 | public string topic() |
将附加到记录的主题。 |
| 2 | public k key() |
将包含在记录中的键。 如果没有这样的键，返回null。 |
| 3 | public v value() |
记录的内容。 |
| 4 | partition() |
记录的分区计数 |
在创建应用程序之前，首先启动zookeeper和kafka代理，然后使用create topic命令在kafka代理中创建自己的主题。 之后，创建一个simpleproducer.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类具有下表中列出的方法。
| 编号 | 方法 | 描述 |
|---|---|---|
| 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) |
参数(pattern)以正则表达式的格式引用订阅模式，而参数listener从订阅模式中获取通知。 |
| 7 | public void as-sign(java.util.list<topicparti-tion> partitions) |
手动将分区列表分配给客户。 |
| 8 | poll() |
获取使用其中一个订阅/分配api指定的主题或分区的数据。 如果在轮询数据之前没有订阅主题，这将返回错误。 |
| 9 | public void commitsync() |
提交最后一次由poll()返回的主题和分区的所有sub-scribed列表的偏移量。 同样的操作应用于commitasyn()。 |
| 10 | public void seek(topicpartition partition, long offset) |
获取消费者将在下一个poll()方法中使用的当前偏移值。 |
| 11 | public void resume() |
恢复暂停的分区。 |
| 12 | public void wakeup() |
唤醒消费者。 |
consumerrecord api用于接收来自kafka集群的记录。 该api由主题名称，分区编号以及指向kafka分区中记录的偏移量组成。 consumerrecord类用于创建具有特定主题名称，分区计数和<key，value>对的消费者记录。 它有以下签名。
public consumerrecord(string topic,int partition, long offset,k key, v value)
参数 -
topic - 从kafka集群收到的消费者记录的主题名称。partition - 主题的分区。key - 记录的关键字，如果不存在关键字,则将返回null。value - 记录内容。consumerrecords api充当consumerrecord的容器。 此api用于保留特定主题的每个分区的consumerrecord列表。 它的构造函数定义如下。
public consumerrecords(java.util.map<topicpartition,java.util.list
<consumer-record>k,v>>> records)
参数
topicpartition - 返回特定主题的分区映射。records - consumerrecord的返回列表。consumerrecords类中定义了以下方法。
| 编号 | 方法 | 描述 |
|---|---|---|
| 1 | public int count() |
所有主题的记录数。 |
| 2 | public set partitions() |
该记录集中包含数据的分区集(如果没有数据返回，则该集为空)。 |
| 3 | public iterator iterator() |
迭代器能够遍历集合，获取或重新移动元素。 |
| 4 | public list records() |
获取给定分区的记录列表。 |
以下列出了consumer客户端api主配置设置的配置设置 -
| 编号 | 设置 | 描述 |
|---|---|---|
| 1 | bootstrap.servers |
经纪人的引导列表。 |
| 2 | group.id |
将个人消费者分配给组。 |
| 3 | enable.auto.commit |
如果值为true，则为偏移启用自动提交，否则不提交。 |
| 4 | auto.commit.interval.ms |
返回更新的消耗偏移量被写入zookeeper的频率。 |
| 5 | session.timeout.ms |
指示在放弃并继续使用消息之前，kafka将等待多少毫秒以等待zookeeper响应请求(读取或写入)。 |
生产者申请步骤在这里保持不变。 首先，启动zookeeper和kafka经纪人。 然后使用simpleconsumer.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 " + 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并发送一些消息给主题。输入如'hello consumer'。得到以类似以下结果 -
subscribed to topic hello-kafka
offset = 3, key = null, value = hello consumer
