一、消息发送
1.1 数据生产流程
数据生产流程图解：
- Producer创建时，会创建⼀个Sender线程并设置为守护线程
- ⽣产消息时，内部其实是异步流程；⽣产的消息先经过 拦截器->序列化器->分区器，然后将消息缓存在缓冲区（该缓冲区也是在Producer创建时创建）
- 批次发送的条件为：缓冲区数据⼤⼩达到
batch.size或者linger.ms达到上限，哪个先达到就算哪个 - 批次发送后，发往指定分区，然后落盘到 broker；如果⽣产者配置了
retrires参数⼤于0并且失败原因允许重试，那么客户端内部会对该消息进⾏重试 - 落盘到broker成功，返回⽣产元数据给⽣产者
- 元数据返回有两种⽅式：⼀种是通过阻塞直接返回，另⼀种是通过回调返回
1.2 必要的参数配置
先来看看我们一般在程序中是怎么配置的：
最常用的配置项：
属性 说明 重要性 bootstrap.servers ⽣产者客户端与broker集群建⽴初始连接需要的broker地址列表，由该初始连接发现Kafka集群中其他的所有broker。该地址列表不需要写全部的Kafka集群中broker的地址，但也不要写⼀个，以防该节点宕机的时候不可⽤。形式为： host1:port1,host2:port2,....
high key.serializer 实现了接⼝org.apache.kafka.common.serialization.Serializer的key序列化类。 high value.serializer 实现了接⼝org.apache.kafka.common.serialization.Serializer的value序列化类。 high acks 该选项控制着已发送消息的持久性。
acks=0
：⽣产者不等待broker的任何消息确认。只要将消息放到了socket的缓冲区，就认为消息已发送。不能保证服务器是否收到该消息，retries设置也不起作⽤，因为客户端不关⼼消息是否发送失败。客户端收到的消息偏移量永远是-1。
acks=1
：leader将记录写到它本地⽇志，就响应客户端确认消息，⽽不等待follower副本的确认。如果leader确认了消息就宕机，则可能会丢失消息，因为follower副本可能还没来得及同步该消息。
acks=all
：leader等待所有同步的副本确认该消息。保证了只要有⼀个同步副本存在，消息就不会丢失。这是最强的可⽤性保证。等价于acks=-1。默认值为1，字符串。可选值：[all, -1, 0, 1] high compression.type ⽣产者⽣成数据的压缩格式。默认是 none
（没有压缩）。允许的值： none gzip snappy lz4
。压缩是对整个消息批次来讲的。消息批的效率也影响压缩的⽐例。消息批越⼤，压缩效率越好。字符串类型的值。默认是 none
。 high retries 设置该属性为⼀个⼤于1的值，将在消息发送失败的时候重新发送消息。该重试与客户端收到异常重新发送并⽆⼆⾄。允许重试但是不设置 max.in.flight.requests.per.connection
为 1，存在消息乱序的可能，因为如果两个批次发送到同⼀个分区，第⼀个失败了重试，第⼆个成功了，则第⼀个消息批在第⼆个消息批后。int类型的值，默认：0，可选值：[0,…,2147483647] high
1.3 拦截器
1.3.1 拦截器介绍
Producer 的拦截器（Interceptor）和 Consumer 的 Interceptor 主要⽤于实现Client端的定制化控制逻辑。
对于Producer⽽⾔，Interceptor使得⽤户在消息发送前以及Producer回调逻辑前有机会对消息做⼀些定制化需求，⽐如修改消息等。同时，Producer允许⽤户指定多个Interceptor按序作⽤于同⼀条消息从⽽形成⼀个拦截链(Interceptor Chain)。Intercetpor 的实现接⼝是 org.apache.kafka.clients.producer.ProducerInterceptor，其定义的⽅法包括：
onSend(ProducerRecord)：该⽅法封装进KafkaProducer.send⽅法中，即运⾏在⽤户主线程中。Producer确保在消息被序列化以计算分区前调⽤该⽅法。⽤户可以在该⽅法中对消息做任何操作，但最好保证不要修改消息所属的topic和分区，否则会影响⽬标分区的计算。onAcknowledgement(RecordMetadata, Exception)：该⽅法会在消息被应答之前或消息发送失败时调⽤，并且通常都是在Producer回调逻辑触发之前。onAcknowledgement运⾏在Producer的IO线程中，因此不要在该⽅法中放⼊很重的逻辑，否则会拖慢Producer的消息发送效率。close：关闭Interceptor，主要⽤于执⾏⼀些资源清理⼯作。
如前所述，Interceptor可能被运⾏在多个线程中，因此在具体实现时⽤户需要 ⾃⾏确保线程安全。另外倘若指定了多个Interceptor，则Producer将按照指定顺序调⽤它们，并仅仅是捕获每个Interceptor可能抛出的异常记录到错误⽇志中⽽⾮在向上传递。这在使⽤过程中要特别留意。
1.3.2 自定义拦截器
自定义拦截器步骤：
- 实现ProducerInterceptor接⼝
- 在KafkaProducer的设置中设置⾃定义的拦截器
自定义拦截器 1：
public class InterceptorOne implements ProducerInterceptor {
private static final Logger LOGGER = LoggerFactory.getLogger(InterceptorOne.class);
@Override
public ProducerRecord onSend(ProducerRecord record) {
System.out.println("拦截器1---go");
// 此处根据业务需要对相关的数据作修改
String topic = record.topic();
Integer partition = record.partition();
Long timestamp = record.timestamp();
Key key = record.key();
Value value = record.value();
Headers headers = record.headers();
// 添加消息头
headers.add("interceptor", "interceptorOne".getBytes());
ProducerRecord newRecord = new ProducerRecord(topic,
partition, timestamp, key, value, headers);
return newRecord;
}
@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
System.out.println("拦截器1---back");
if (exception != null) {
// 如果发⽣异常，记录⽇志中
LOGGER.error(exception.getMessage());
}
}
@Override
public void close() {
}
@Override
public void configure(Map configs) {
}
}
照着 拦截器 1 再加两个拦截器。
生产者
public class MyProducer1 {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
Map configs = new HashMap<>();
// 设置连接Kafka的初始连接⽤到的服务器地址
// 如果是集群，则可以通过此初始连接发现集群中的其他broker
configs.put("bootstrap.servers", "192.168.0.102:9092");
// 设置key的序列化器
configs.put("key.serializer", IntegerSerializer.class);
// 设置⾃定义的序列化类
configs.put("value.serializer", UserSerializer.class);
// 设置自定义分区器
configs.put("partitioner.class", "com.mfc.config.MyPartitioner");
// 设置拦截器
configs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
"com.mfc.interceptor.InterceptorOne,"
+ "com.mfc.interceptor.InterceptorTwo,"
+ "com.mfc.interceptor.InterceptorThree");
KafkaProducer producer = new KafkaProducer<>(configs);
User user = new User();
user.setUserId(1001);
user.setUsername("阿彪");
// ⽤于封装Producer的消息
ProducerRecord record = new ProducerRecord<>(
"topic_1", // 主题名称
0, // 分区编号
user.getUserId(), // 数字作为key
user // user 对象作为value
);
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception e) {
if (e == null) {
System.out.println("消息发送成功：" + metadata.topic() + "\t"
+ metadata.partition() + "\t"
+ metadata.offset());
} else {
System.out.println("消息发送异常");
}
}
});
// 关闭⽣产者
producer.close();
}
}
1.4 序列化器
1.4.1 Kafka 自带序列化器
Kafka使⽤ org.apache.kafka.common.serialization.Serializer接⼝⽤于定义序列化器，将泛型指定类型的数据转换为字节数组。
package org.apache.kafka.common.serialization;
import java.io.Closeable;
import java.util.Map;
/**
将对象转换为byte数组的接⼝
该接⼝的实现类需要提供⽆参构造器
@param 从哪个类型转换
*/
public interface Serializer extends Closeable {
/*
类的配置信息
@param configs key/value pairs
@param isKey key的序列化还是value的序列化
*/
void configure(Map var1, boolean var2);
/*
将对象转换为字节数组
@param topic 主题名称
@param data 需要转换的对象
@return 序列化的字节数组
*/
byte[] serialize(String var1, T var2);
/*
关闭序列化器
该⽅法需要提供幂等性，因为可能调⽤多次。
*/
void close();
}
系统提供了该接⼝的⼦接⼝以及实现类：
org.apache.kafka.common.serialization.ByteArraySerializer
org.apache.kafka.common.serialization.ByteBufferSerializer
org.apache.kafka.common.serialization.BytesSerializer
org.apache.kafka.common.serialization.DoubleSerializer
org.apache.kafka.common.serialization.FloatSerializer
org.apache.kafka.common.serialization.IntegerSerializer
org.apache.kafka.common.serialization.StringSerializer
org.apache.kafka.common.serialization.LongSerializer
org.apache.kafka.common.serialization.ShortSerializer
1.4.2 自定义序列化器
数据的序列化⼀般⽣产中使⽤ avro。
⾃定义序列化器需要实现 org.apache.kafka.common.serialization.Serializer<t></t> 接⼝，并实现其中的 serialize⽅法。
实体类
public class User {
private Integer userId;
private String username;
// set、get方法省略
}
自定义序列化器
public class UserSerializer implements Serializer {
@Override
public void configure(Map map, boolean b) {
// do Nothing
}
@Override
public byte[] serialize(String topic, User user) {
try {
// 如果数据是null，则返回null
if (user == null) return null;
Integer userId = user.getUserId();
String username = user.getUsername();
int length = 0;
byte[] bytes = null;
if (null != username) {
bytes = username.getBytes("utf-8");
length = bytes.length;
}
ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + length);
buffer.putInt(userId);
buffer.putInt(length);
buffer.put(bytes);
return buffer.array();
} catch (UnsupportedEncodingException e) {
throw new SerializationException("序列化数据异常");
}
}
@Override
public void close() {
// do Nothing
}
}
生产者：
public class MyProducer1 {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
Map configs = new HashMap<>();
// 设置连接Kafka的初始连接⽤到的服务器地址
// 如果是集群，则可以通过此初始连接发现集群中的其他broker
configs.put("bootstrap.servers", "192.168.0.102:9092");
// 设置key的序列化器
configs.put("key.serializer", IntegerSerializer.class);
// 设置⾃定义的序列化类
configs.put("value.serializer", UserSerializer.class);
KafkaProducer producer = new KafkaProducer<>(configs);
User user = new User();
user.setUserId(1001);
user.setUsername("阿彪");
// ⽤于封装Producer的消息
ProducerRecord record = new ProducerRecord<>(
"topic_1", // 主题名称
0, // 分区编号
user.getUserId(), // 数字作为key
user // user 对象作为value
);
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception e) {
if (e == null) {
System.out.println("消息发送成功：" + metadata.topic() + "\t"
+ metadata.partition() + "\t"
+ metadata.offset());
} else {
System.out.println("消息发送异常");
}
}
});
// 关闭⽣产者
producer.close();
}
}
1.5 分区器
1.5.1 Kafka 自带分区器
默认（ DefaultPartitioner）分区计算：
- 如果record提供了分区号，则使⽤record提供的分区号
- 如果record没有提供分区号，则使⽤key的序列化后的值的hash值对分区数量取模
- 如果record没有提供分区号，也没有提供key，则使⽤轮询的⽅式分配分区号。
- 会⾸先在可⽤的分区中分配分区号
- 如果没有可⽤的分区，则在该主题所有分区中分配分区号。
看一下kafka的生产者（ KafkaProducer）源码：
再看Kafka自带的默认分区器（ DefaultPartitioner）:
默认的分区器实现了 Partitioner 接口，先看一下接口：
public interface Partitioner extends Configurable, Closeable {
/**
* 为指定的消息记录计算分区值
*
* @param topic 主题名称
* @param key 根据该key的值进⾏分区计算，如果没有则为null
* @param keyBytes key的序列化字节数组，根据该数组进⾏分区计算。如果没有key，则为null
* @param value 根据value值进⾏分区计算，如果没有，则为null
* @param valueBytes value的序列化字节数组，根据此值进⾏分区计算。如果没有，则为null
* @param cluster 当前集群的元数据
*/
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
/**
* 关闭分区器的时候调⽤该⽅法
*/
public void close();
}
1.5.2 自定义分区器
如果要⾃定义分区器，则需要
- ⾸先开发Partitioner接⼝的实现类
- 在KafkaProducer中进⾏设置：
configs.put("partitioner.class", "xxx.xx.Xxx.class")
实现Partitioner接⼝⾃定义分区器：
public class MyPartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
return 0;
}
@Override
public void close() {
}
@Override
public void configure(Map configs) {
}
}
然后在⽣产者中配置：
二、消息发送原理
原理图解：
由上图可以看出： KafkaProducer 有两个基本线程：
- 主线程：负责消息创建，拦截器，序列化器，分区器等操作，并将消息追加到消息收集器
RecoderAccumulator中； - 消息收集器
RecoderAccumulator为每个分区都维护了⼀个Deque<producerbatch></producerbatch>类型的双端队列。 ProducerBatch可以理解为是ProducerRecord的集合，批量发送有利于提升吞吐量，降低⽹络影响；- 由于⽣产者客户端使⽤
java.io.ByteBuffer在发送消息之前进⾏消息保存，并维护了⼀个BufferPool实现ByteBuffer的复⽤；该缓存池只针对特定⼤⼩（batch.size指定）的ByteBuffer进⾏管理，对于消息过⼤的缓存，不能做到重复利⽤。 - 每次追加⼀条
ProducerRecord消息，会寻找/新建对应的双端队列，从其尾部获取⼀个ProducerBatch，判断当前消息的⼤⼩是否可以写⼊该批次中。若可以写⼊则写⼊；若不可以写⼊，则新建⼀个ProducerBatch，判断该消息⼤⼩是否超过客户端参数配置batch.size的值，不超过，则以batch.size建⽴新的ProducerBatch，这样⽅便进⾏缓存重复利⽤；若超过，则以计算的消息⼤⼩建⽴对应的ProducerBatch，缺点就是该内存不能被复⽤了。 Sender线程：- 该线程从消息收集器获取缓存的消息，将其处理为
<node, list<producerbatch></node,>的形式， Node 表示集群的broker节点。 - 进⼀步将
<node, list<producerbatch></node,>转化为<node, request></node,>形式，此时才可以向服务端发送数据。 - 在发送之前，
Sender线程将消息以Map<nodeid, deque<request>> </nodeid,>的形式保存到InFlightRequests中进⾏缓存，可以通过其获取leastLoadedNode,即当前Node中负载压⼒最⼩的⼀个，以实现消息的尽快发出。
三、更多生产者参数配置
参数名称 描述 retry.backoff.ms 在向⼀个指定的主题分区重发消息的时候，重试之间的等待时间。
⽐如3次重试，每次重试之后等待该时间⻓度，再接着重试。在⼀些失败的场景，避免了密集循环的重新发送请求。
long型值，默认100。可选值：[0,…] retries retries重试次数
当消息发送出现错误的时候，系统会重发消息。
跟客户端收到错误时重发⼀样。
如果设置了重试，还想保证消息的有序性，需要设置 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1
否则在重试此失败消息的时候，其他的消息可能发送成功了 request.timeout.ms 客户端等待请求响应的最⼤时⻓。如果服务端响应超时，则会重发请求，除⾮达到重试次数。该设置应该⽐ replica.lag.time.max.ms (a broker configuration)
要⼤，以免在服务器延迟时间内重发消息。int类型值，默认：30000，可选值：[0,…] interceptor.classes 在⽣产者接收到该消息，向Kafka集群传输之前，由序列化器处理之前，可以通过拦截器对消息进⾏处理。
要求拦截器类必须实现 org.apache.kafka.clients.producer.ProducerInterceptor
接⼝。默认没有拦截器。
Map<string, object> configs</string,>
中通过List集合配置多个拦截器类名。 acks 默认值：all。
acks=0：
⽣产者不等待broker对消息的确认，只要将消息放到缓冲区，就认为消息已经发送完成。
该情形不能保证broker是否真的收到了消息，retries配置也不会⽣效。发送的消息的返回的消息偏移量永远是-1。
acks=1
表示消息只需要写到主分区即可，然后就响应客户端，⽽不等待副本分区的确认。
在该情形下，如果主分区收到消息确认之后就宕机了，⽽副本分区还没来得及同步该消息，则该消息丢失。
acks=all
⾸领分区会等待所有的ISR副本分区确认记录。
该处理保证了只要有⼀个ISR副本分区存活，消息就不会丢失。
这是Kafka最强的可靠性保证，等效于acks=-1 batch.size 当多个消息发送到同⼀个分区的时候，⽣产者尝试将多个记录作为⼀个批来处理。批处理提⾼了客户端和服务器的处理效率。
该配置项以字节为单位控制默认批的⼤⼩。
所有的批⼩于等于该值。
发送给broker的请求将包含多个批次，每个分区⼀个，并包含可发送的数据。
如果该值设置的⽐较⼩，会限制吞吐量（设置为0会完全禁⽤批处理）。如果设置的很⼤，⼜有⼀点浪费内存，因为Kafka会永远分配这么⼤的内存来参与到消息的批整合中。 client.id ⽣产者发送请求的时候传递给broker的id字符串。
⽤于在broker的请求⽇志中追踪什么应⽤发送了什么消息。
⼀般该id是跟业务有关的字符串。 compression.type ⽣产者发送的所有数据的压缩⽅式。默认是none，也就是不压缩。
⽀持的值：none、gzip、snappy和lz4。
压缩是对于整个批来讲的，所以批处理的效率也会影响到压缩的⽐例。 send.buffer.bytes TCP发送数据的时候使⽤的缓冲区（SO_SNDBUF）⼤⼩。如果设置为0，则使⽤操作系统默认的。 buffer.memory ⽣产者可以⽤来缓存等待发送到服务器的记录的总内存字节。如果记录的发送速度超过了将记录发送到服务器的速度，则⽣产者将阻塞 max.block.ms
的时间，此后它将引发异常。此设置应⼤致对应于⽣产者将使⽤的总内存，但并⾮⽣产者使⽤的所有内存都⽤于缓冲。⼀些额外的内存将⽤于压缩（如果启⽤了压缩）以及维护运⾏中的请求。long型数据。默认值：33554432，可选值：[0,…] connections.max.idle.ms 当连接空闲时间达到这个值，就关闭连接。long型数据，默认：540000 linger.ms ⽣产者在发送请求传输间隔会对需要发送的消息进⾏累积，然后作为⼀个批次发送。⼀般情况是消息的发送的速度⽐消息累积的速度慢。有时客户端需要减少请求的次数，即使是在发送负载不⼤的情况下。该配置设置了⼀个延迟，⽣产者不会⽴即将消息发送到broker，⽽是等待这么⼀段时间以累积消息，然后将这段时间之内的消息作为⼀个批次发送。该设置是批处理的另⼀个上限：⼀旦批消息达到了 batch.size
指定的值，消息批会⽴即发送，如果积累的消息字节数达不到 batch.size
的值，可以设置该毫秒值，等待这么⻓时间之后，也会发送消息批。该属性默认值是0（没有延迟）。如果设置 linger.ms=5
，则在⼀个请求发送之前先等待5ms。long型值，默认：0，可选值：[0,…] max.block.ms 控制 KafkaProducer.send() KafkaProducer.partitionsFor()
阻塞的时⻓。当缓存满了或元数据不可⽤的时候，这些⽅法阻塞。在⽤户提供的序列化器和分区器的阻塞时间不计⼊。long型值，默认：60000，可选值：[0,…] max.request.size 单个请求的最⼤字节数。该设置会限制单个请求中消息批的消息个数，以免单个请求发送太多的数据。服务器有⾃⼰的限制批⼤⼩的设置，与该配置可能不⼀样。int类型值，默认1048576，可选值：[0,…] partitioner.class 实现了接⼝ org.apache.kafka.clients.producer.Partitioner
的分区器实现类。默认值为： org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes TCP接收缓存（SO_RCVBUF），如果设置为-1，则使⽤操作系统默认的值。int类型值，默认32768，可选值：[-1,…] security.protocol 跟broker通信的协议：PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.
string类型值，默认：PLAINTEXT max.in.flight.requests.per.connection 单个连接上未确认请求的最⼤数量。达到这个数量，客户端阻塞。如果该值⼤于1，且存在失败的请求，在重试的时候消息顺序不能保证。
int类型值，默认5。可选值：[1,…] reconnect.backoff.max.ms 对于每个连续的连接失败，每台主机的退避将成倍增加，直⾄达到此最⼤值。在计算退避增量之后，添加20％的随机抖动以避免连接⻛暴。
long型值，默认1000，可选值：[0,…] reconnect.backoff.ms 尝试重连指定主机的基础等待时间。避免了到该主机的密集重连。该退避时间应⽤于该客户端到broker的所有连接。
long型值，默认50。可选值：[0,…]
Original: https://www.cnblogs.com/fanchengmeng/p/16282290.html
Author: 下半夜的风
Title: Kafka 生产者解析
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/571924/
转载文章受原作者版权保护。转载请注明原作者出处！