Kafka生产者—消息发送流程，同步、异步发送API

生产者消息发送流程

发送原理

Kafka的Producer发送消息采用的是异步发送的方式。

在消息发送的过程中，涉及到了两个线程:main线程和Sender线程，以及一个线程共享变量:RecordAccumulator。

①main线程中创建了一个双端队列RecordAccumulator，将消息发送给RecordAccumulator。

②Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。

batch.size: 只有数据积累到batch.size之后，sender才会发送数据。默认16k

linger.ms: 如果数据迟迟未达到batch.size，sender等待linger.ms设置的时间到了之后就会发送数据。单位ms，默认值是0ms，表示没有延迟。

0:生产者发送过来的数据，不需要等数据落盘应答。1:生产者发送过来的数据，Leader收到数据后应答。-1 ( al1） :生产者发送过来的数据，Leader和和ISR队列里面的所有节点收齐数据后应答。-1和al1等价。

生产者重要参数列表

bootstrap.servers: 生产者连接集群所需的broker地址清单。可以设置1个或者多个，中间用逗号隔开。生产者从给定的broker里查找到其他broker信息。

key.serializer、 value.serializer: 指定发送消息的key和value的序列化类型。要写全类名。（反射获取）

buffer.memory: RecordAccumulator缓冲区总大小，默认32m。

batch.size: 缓冲区一批数据最大值，默认16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。

linger.ms: 如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。单位ms，默认值是0ms，表示没有延迟。生产环境建议该值大小为5-100ms之间。

acks:

0：生产者发送过来的数据，不需要等数据落盘应答。

1：生产者发送过来的数据，Leader数据落盘后应答。

-1（all）：生产者发送过来的数据，Leader和isr队列里面的所有节点数据都落盘后应答。默认值是-1

max.in.flight.requests.per.connection: 允许最多没有返回ack的次数，默认为5，开启幂等性要保证该值是 1-5的数字。

Retries（重试）: 当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是int最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。

retry.backoff.ms: 两次重试之间的时间间隔，默认是100ms。

enable.idempotence: 是否开启幂等性，默认true，开启幂等性。

compression.type 生产者发送的所有数据的压缩方式。默认是none，不压缩。 支持压缩类型：none、gzip、snappy、lz4和zstd。

异步发送API

普通异步发送

需求：创建Kafka生产者，采用异步的方式发送到Kafka broker

异步发送流程如下：

batch.size: 只有数据积累到batch.size之后，sender才会发送数据。默认16k

linger.ms: 如果数据迟迟未达到batch.size，sender等待lingerms设置的时间到了之后就会发送数据。单位ms，默认值是Oms，表示没有延迟。

代码编写

1）创建工程kafka-demo

2）导入依赖

org.apache.kafka

kafka-clients

3.0.0

3）创建包名：com.taohua.kafka.producer

4）编写代码：不带回调函数的API

package com.taohua.kafka.producer;

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

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

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

import java.util.Properties;

public class CustomProducer {

public static void main(String[] args) throws InterruptedException {

// 1. 创建kafka生产者的配置对象

Properties properties = new Properties();

// 2. 给kafka配置对象添加配置信息

properties.put("bootstrap.servers","hadoop102:9092");

// key,value序列化

properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 3. 创建kafka生产者对象

KafkaProducer kafkaProducer = new KafkaProducer(properties);

// 4. 调用send方法,发送消息

for (int i = 0; i < 10; i++) {

kafkaProducer.send(new ProducerRecord<>("first","kafka" + i));

}

// 5. 关闭资源

kafkaProducer.close();

}

}

5）测试：

在hadoop102上开启kafka消费者

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

1

在IDEA中执行上述代码，观察hadoop102消费者输出

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

kafka0

kafka1

kafka2

kafka3

……

带回调函数的异步发送

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

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

·如果Exception为null，说明消息发送成功，

·如果Exception不为null，说明消息发送失败。

带回调函数的异步调用发送流程

batch.size: 只有数据积累到batch.size之后，sender才会发送数据。默认16k

linger.ns: 如果数据迟迟未达到batch.size，sender等待linger:ms设置的时间到了之后就会发送数据。单位ms，默认值是Oms，表示没有延迟。

编写代码：带回调函数的生产者

package com.taohua.kafka.producer;

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

import java.util.Properties;

public class CustomProducerCallback {

public static void main(String[] args) throws InterruptedException {

// 1. 创建kafka生产者的配置对象

Properties properties = new Properties();

// 2. 给kafka配置对象添加配置信息

properties.put("bootstrap.servers", "hadoop102:9092");

properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

// key,value序列化(必须)

properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 3. 创建kafka生产者对象

KafkaProducer kafkaProducer = new KafkaProducer(properties);

// 4. 调用send方法,发送消息

for (int i = 0; i < 10; i++) {

// 添加回调

kafkaProducer.send(new ProducerRecord<>("first", "kafka" + i), new Callback() {

// 该方法在Producer收到ack时调用，为异步调用

@Override

public void onCompletion(RecordMetadata metadata, Exception exception) {

if (exception == null)

// 没有异常,输出信息到控制台

System.out.println("主题"+recordMetadata.topic() +", 分区："+recordMetadata.partition()+", 偏移量："+recordMetadata.offset());

}

});

}

// 5. 关闭资源

kafkaProducer.close();

}

}

测试

1）在hadoop102上开启kafka消费者

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

1

2）在IDEA中执行代码，观察hadoop102消费者输出

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

kafka0

kafka1

kafka2

……

3）在IDEA控制台观察回调信息

主题first, 分区：0, 偏移量：10

主题first, 分区：0, 偏移量：11

主题first, 分区：0, 偏移量：12

主题first, 分区：0, 偏移量：13

主题first, 分区：0, 偏移量：14

主题first, 分区：0, 偏移量：15

主题first, 分区：0, 偏移量：16

主题first, 分区：0, 偏移量：17

主题first, 分区：0, 偏移量：18

主题first, 分区：0, 偏移量：19

……

同步发送API

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

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

同步发送流程示意图如下：

batch.size: 只有数据积累到batch.size之后，sender才会发送数据。默认16k

linger.ns: 如果数据迟迟未达到batch.size，sender等待linger:ms设置的时间到了之后就会发送数据。单位ms，默认值是Oms，表示没有延迟。

编写代码：同步发送消息的生产者

package com.atguigu.kafka.producer;

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

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

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

import java.util.Properties;

import java.util.concurrent.ExecutionException;

public class ConsumerProducerSync {

public static void main(String[] args) throws InterruptedException, ExecutionException {

// 1. 创建kafka生产者的配置对象

Properties properties = new Properties();

// 2. 给kafka配置对象添加配置信息

//properties.put("bootstrap.servers","hadoop102:9092");

properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");

// key,value序列化(必须)

properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 3. 创建kafka生产者对象

KafkaProducer kafkaProducer = new KafkaProducer(properties);

// 4. 调用send方法,发送消息

for (int i = 0; i < 10; i++) {

// 同步发送

kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();

}

// 5. 关闭资源

kafkaProducer.close();

}

}

测试

1）在hadoop102上开启kafka消费者

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

1

2）在IDEA中执行代码，观察hadoop102消费者的消费情况

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

kafka0

kafka1

kafka2

……