同步发送API

https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients

dependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactIdversion3.6.0/version

/dependency

org.apache.kafka.clients.producer.KafkaProducer;

import

org.apache.kafka.clients.producer.ProducerConfig;

import

org.apache.kafka.clients.producer.ProducerRecord;

import

org.apache.kafka.clients.producer.RecordMetadata;

import

org.apache.kafka.common.serialization.StringSerializer;import

import

java.util.concurrent.ExecutionException;/***

author

给kafka配置对象添加配置信息properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,

key,value序列化properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,

StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,

StringSerializer.class.getName());//

创建kafka生产者对象KafkaProducerString,

String

i)).get();System.out.println(art.offset());System.out.println(over

i);}//

回调函数callback会在producer收到ack时调用为异步调用。

该方法有两个参数分别是RecordMetadata元数据信息和Exception异常信息。

如果Exception为null说明消息发送成功如果Exception不为null说明消息发送失败

Code

org.apache.kafka.clients.producer.*;

import

org.apache.kafka.common.serialization.StringSerializer;import

import

java.util.concurrent.ExecutionException;/***

author

给kafka配置对象添加配置信息properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,

key,value序列化properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,

StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,

StringSerializer.class.getName());//

创建kafka生产者对象KafkaProducerString,

String

该方法在Producer收到ack时调用为异步调用kafkaProducer.send(new

ProducerRecord(art,

没有异常,输出信息到控制台System.out.println(主题

偏移量

recordMetadata.offset());});}//

控制台

同步发送的意思就是一条消息发送之后会阻塞当前线程直至返回ack。

由于send方法返回的是一个Future对象根据Futrue对象的特点我们也可以实现同步发送的效果只需在调用Future对象的get方发即可。

package

org.apache.kafka.clients.producer.KafkaProducer;

import

org.apache.kafka.clients.producer.ProducerConfig;

import

org.apache.kafka.clients.producer.ProducerRecord;

import

org.apache.kafka.common.serialization.StringSerializer;import

import

java.util.concurrent.ExecutionException;/***

author

给kafka配置对象添加配置信息properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,

key,value序列化properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,

StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,

StringSerializer.class.getName());//

创建kafka生产者对象KafkaProducerString,

String

通过Future接口的get实现同步阻塞kafkaProducer.send(new

ProducerRecord(art,