一、新建一个项目
本文我们为大家介绍apache kafka生产者同步及异步发送数据三种方式，基于java项目实现。为了方便应用，我们新建一个java的maven项目
引入kafka 的Java客户端依赖，同时假如JUnit5单元测试依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.6.0</version>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<version>RELEASE</version>
<scope>test</scope>
</dependency>
二、设置Producer客户端参数
ProducerConfig：设置生产者客户端的一系列配置参数。
public class kafkaProducerTest {
private Properties props;
@BeforeEach
public void preTest(){
props = new Properties();
//kafka broker服务列表
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.137.10:9092");
//可靠性确认应答参数
props.put(ProducerConfig.ACKS_CONFIG, "all");
//发送失败，重新尝试的次数
props.put(ProducerConfig.RETRIES_CONFIG, 1);
//生产者数据key的格式
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//生产者数据value的格式
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
}
}
三、构建消息对象
kafka 的ProducerRecord就是我们要发送的消息，里面包含了很多的信息，主要有以下几个构造函数，可以更具自己的需求选择合适的构造函数。
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {}
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {}
public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers) {}
public ProducerRecord(String topic, Integer partition, K key, V value) {}
puic ProducerRecord(String topic, K key, V value) {}
public ProducerRecord(String topic, V value) {}
- topic：Topic主题的的名称
- partition：主题的分区编号，编号从0开始。表示消息数据指定发送到该分区中
- timestamp：时间戳，一般默认当前时间戳
- key：消息的键，可以是不同数据类型，但是通常是String。具有相同key的消息被发往同一个分区，也就是说具有相同key的消息可以保证数据有序性。
- data：消息的数据，可以是不同数据类型
- headers：消息的自定义标签信息
四、三种数据发送方式
Kafka中有三种发送消息的方式：
- 只发不管结果（fire-and-forget）：只调用接口发送消息到Kafka服务器，但不管成功写入与否。由于Kafka是高可用的，因此大部分情况下消息都会写入，但在异常情况下会丢消息。
- 同步发送（Synchronous send）：调用send()方法返回一个Future对象，我们可以使用它的get()方法来判断消息发送成功与否。会阻塞当前线程，执行效率低，但可靠性最高。
- 异步发送（Asynchronous send）：调用send()时提供一个回调方法，不会阻塞当前线程，当接收到broker结果后回调此方法。可靠性不如同步发送，可能因为等问题不保证100%被回调。
4.1.不带回调函数
KafkaProducer#send(ProducerRecord<K,V>)
- KafkaProducer：生产者对象，用来发送数据
- ProducerRecord：每条数据都要封装成一个ProducerRecord 对象
//没有回调函数的调用方法
@Test
public void noCallback() {
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 20; i++) {
producer.send(
new ProducerRecord<>(
"producer_test",
Integer.toString(i),
"noCallback value:" + i)
);
}
producer.close();
}
4.2.带回调函数
KafkaProducer#send(ProducerRecord<K,V>, Callback)
使用方法
@Test
public void withCallback() {
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("producer_test", "withCallback value:" ),
(metadata, exception) -> { //回调函数
if (null == exception) {
System.out.println("消息数据发送成功 ->" + metadata.offset());
} else {
exception.printStackTrace();
}
});
producer.close();
}
回调函数会在producer收到ack之前异步调用，该方法有两个参数：RecordMetadata、Exception。这两个参数是互斥的关系，即：
- 如果Exception为null，则消息发送成功，此时RecordMetadata必定不为null。
- 消息发送异常时，RecordMetadata为null，而exception不为null。消息发送失败会自动重试，不需在回调函数中手动重试。重试次数由参数retries设置来设定，重试次数达到上线后，仍然发送失败才会有exception的存在。
4.3.同步发送
KafkaProducer的send方法都是异步的，由上面异步的例子可以看出，producer的send方法返回对象是Future类型，因此可以通过调用Future对象的get()方法阻塞，等待发送结果的响应，从而达到同步发送消息的效果。这里的同步是指，一条消息发送后会阻塞当前线程，直至返回ack消息。
因此与异步代码区别只有一处（get方法）：
//同步数据send
@Test
public void syncSend() throws ExecutionException, InterruptedException {
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
RecordMetadata metadata = producer.send(new ProducerRecord<>(
"producer_test",
"syncSend value" ))
.get();
if(metadata != null && metadata.hasOffset()){
System.out.println("数据同步发送确认成功 ->" + metadata.offset());
}
producer.close();
}