开局一张图,其他全靠吹
发送消息流程如下:
1.初始化流程
-
指定bootstrap.servers,地址的格式为
host:port
。它会连接bootstrap.servers参数指定的所有Broker,Producer启动时会发起与这些Broker的连接。因此,如果你为这个参数指定了1000个Broker连接信息,那么很遗憾,你的Producer启动时会首先创建与这1000个Broker的TCP连接。- 在实际使用过程中,我并不建议把集群中所有的Broker信息都配置到bootstrap.servers中,通常你指定3~4台就足以了。因为Producer一旦连接到集群中的任一台Broker,就能拿到整个集群的Broker信息,故没必要为bootstrap.servers指定所有的Broker。
props.put("bootstrap.servers", "localhost:9092");
-
指定Key和Value的序列化方式。
-
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-
-
指定acks配置,默认值是all(版本3.x)
props.put("acks", "all");
- 设置为0,表示生产端发送消息后立即返回,不等待broker端的响应结果。通常此时生产端吞吐量最高,消息发送的可靠性最低。
- 设置为1,表示leader副本成功写入PageCache就会响应Producer,而无需等待ISR(同步副本)集合中的其他副本写入成功。这种方案提供了适当的持久性,保证了一定的吞吐量。
- 设置成all或-1,表示不仅要等leader副本成功写入,还要求ISR中的其他副本成功写入,才会响应Producer。这种方案提供了最高的持久性,但也提供了最差的吞吐量。
-
producer = new KafkaProducer<>(props);
- 从配置中获取必要的参数,如
transactionalId
和clientId
。 - 根据
clientId
创建日志记录上下文(LogContext),用于日志记录。 - 配置度量(Metrics)相关信息,包括度量标签、度量配置、度量报告器等。
- 创建度量上下文(MetricsContext)和度量实例(Metrics)。
- 初始化分区器(Partitioner)。
- 配置并初始化键(key)和值(value)的序列化器(Serializer)。
- 配置并初始化拦截器(Interceptors)。
- 配置集群资源监听器(ClusterResourceListeners)。
- 设置最大请求大小(maxRequestSize)、内存大小(totalMemorySize)和压缩类型(compressionType)等参数。
- 配置最大阻塞时间(maxBlockTimeMs)和交付超时时间(deliveryTimeoutMs)。
- 初始化API版本(apiVersions)和事务管理器(transactionManager)。
- 创建记录累加器(RecordAccumulator),用于累积记录以进行批量发送。
- 解析并验证引导服务器地址(addresses)。
- 如果提供了元数据(metadata),则使用提供的元数据,否则创建新的元数据实例,并通过引导服务器地址进行引导。
- 初始化错误度量传感器(errors)。
- 创建并启动IO线程(ioThread)来处理消息发送。
- 注册应用程序信息,用于JMX度量和监控。
- 如果在初始化过程中发生任何错误,将调用关闭方法以避免资源泄漏,并向上抛出Kafka异常。
- 从配置中获取必要的参数,如
2.发送消息流程
在消息发送的过程中,涉及到了两个线程——
main
线程和Sender
线程。在main
线程中创建了一个双端队列RecordAccumulator
。main
线程将消息发送给RecordAccumulator
,Sender
线程不断从RecordAccumulator
中拉取消息发送到Kafka Broker
。
-
构造消息记录ProducerRecord 对象,对象包含了四个属性:Topic,partition,key,value;topic 和 value 是必须的,key 和 partition 是可选的。
-
同步获取Kafka集群信息(Cluster)。
- 如果缓存有,并且分区没有超过指定分区范围则返回缓存
- 否则触发更新,等待从broker获取新的元数据信息
- 默认强制拉取时间是metadata.max.age.ms: 5分钟
-
使用键序列化器(keySerializer)将消息的键序列化为字节数组,使用值序列化器(valueSerializer)将消息的值序列化为字节数组。
-
计算数据发送到那个分区,如果指定了 key,那么相同 key 的消息会发往同一个分区,如果实现了自定义分区器,那么就会走自定义分区器进行分区路由。
- 如果有Key值,则使用Key值的Hash值来分配分区
murmurhash(key) % 主题分区总数
。 - 老版本:如果没有key值,则以Round-Robin的方式分配分区。
- 新版本:如果没有key值,则以粘性分区的方式分配分区
- 如果有Key值,则使用Key值的Hash值来分配分区
-
创建一个
TopicPartition
对象,表示要发送消息的主题和分区。 -
判断消息的大小是否超过了我们设置的阈值。
-
异步发送时,给每一条消息都绑定他的回调函数
-
把消息放入记录累加器(accumulator)(32M的一个内存),然后有accumulator把消息封装成为一个批次一个批次的去发送。
-
如果批次满了或者新创建出来一个批次, 唤醒sender线程,他才是真正发送数据的线程,发送的时候并不是来一个消息就发送一个消息,这样的话吞吐量比较低,并且频繁的进行网络请求。消息是按照批次来发送的或者等待时间来发的的.
-
Leader Broker接收到消息写入到PageCache,当Producer的acks设置为"all"时,这意味着Producer会等待所有ISR(In-Sync Replicas,即同步副本)都成功确认消息之后才会认为消息发送成功。
-
Leader Partition接收消息:每个Partition都有一个Leader Broker,该Leader Broker负责接收所有消息并处理副本同步。Leader Partition会接收Producer发送的消息。
-
消息复制到Follower副本:Leader Partition会将消息复制到ISR中的Follower副本(In-Sync Replicas),这些副本是与Leader保持同步的副本。Kafka允许配置多个ISR,以提高可用性。
-
等待Follower副本确认:Leader Partition会等待ISR中的Follower副本确认已成功复制消息。这些确认信息包括副本在哪个Offset处复制消息。
-
Leader确认消息:一旦ISR中的所有Follower副本都确认了消息,Leader Partition会向Producer发送确认消息。这表示消息已经成功写入Leader Partition并且已在ISR中的所有Follower副本中成功复制。
-
Producer接收确认:Producer收到来自Kafka的确认消息,这时候Producer认为消息发送成功。
-
定期提交消息到磁盘:Leader Partition会定期将已接收的消息写入磁盘以确保持久性,以防Broker故障。
-
Follower副本的同步:Follower副本会定期从Leader Partition拉取消息,确保与Leader保持同步。如果Follower副本无法赶上Leader,它可能会被认为是“失去同步”,不再被视为ISR的一部分。
示例
Properties props = new Properties();
// 配置broker列表,一般配置3-5个即可,如果你有100个broker,不需要全部配置,能连上一个就行
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");
// 持久化机制参数
props.put(ProducerConfig.ACKS_CONFIG, "-1");
// 发送失败会重试发送失败会重试,默认重试间隔100ms,重试能保证消息发送的可靠性,但是也可能造成消息重复发送,比如网络抖动,所以需要在 接收者那边做好消息接收的幂等性处理
// 注意:消息发送失败会自动重试,不需要我们在回调函数中手动重试
props.put(ProducerConfig.RETRIES_CONFIG, 3);
// 设置重试间隔
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
// 设置发送消息的本地缓冲区大小 32MB
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
// kafka本地线程会从缓冲区取数据,批量发送到broker,
// 设置批量发送消息的大小,默认值是16384,即16kb,就是说一个batch满了16kb就发送出去
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
// 默认值是0,意思就是消息必须立即被发送,但这样会影响性能
// 一般设置100毫秒左右,就是说这个消息发送完后会进入本地的一个batch,如果100毫秒内,这个batch满了16kb就会随batch一起被发送出去
// 如果100毫秒内,batch没满,那么也必须把消息发送出去,不能让消息的发送延迟时间太长
props.put(ProducerConfig.LINGER_MS_CONFIG, 100);
// 把发送的key从字符串序列化为字节数组
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//把发送消息value从字符串序列化为字节数组
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
Producer<String, String> producer = new KafkaProducer<String, String>(props);
for (int i = 1; i <= msgNum; i++) {
Order order = new Order(i, 100 + i, 1, 1000.00);
// 1.指定发送分区
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
, 0, order.getOrderId().toString(), JSON.toJSONString(order));
// 2.未指定发送分区,指定Key 具体发送的分区计算公式:hash(key) % partitionNum
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
, order.getOrderId().toString(), JSON.toJSONString(order));
// 3.既没有指定分区,也没有指定Key Kafka采用Sticky Partition(黏性分区器),会随机选择一个分区,并尽可能一直使用该分区,
// 待该分区的batch已满或者或者linger.ms设置的时间到,Kafka再随机一个分区进行使用(和上一次的分区不同)
// 4.自定义分区器
// 同步发送:等待消息发送成功的同步阻塞方法
RecordMetadata metadata = producer.send(producerRecord).get();
System.out.println("同步方式发送消息结果:" + "topic-" + metadata.topic() + "|partition-"
+ metadata.partition() + "|offset-" + metadata.offset());*/
// 异步回调方式发送
producer.send(producerRecord, new Callback() {
// 回调函数会在 producer 收到 ack 时调用,为异步调用,该方法有两个参数,分别是元数据信息(RecordMetadata)和异常信息(Exception),
// 如果 Exception 为 null,说明消息发送成功,如果 Exception 不为 null,说明消息发送失败
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
System.err.println("发送消息失败:" + exception.getStackTrace());
}
if (metadata != null) {
System.out.println("异步方式发送消息结果:" + "topic-" + metadata.topic() + "|partition-"
+ metadata.partition() + "|offset-" + metadata.offset());
}
}
});
生产者发送消息三大模式
- 发后即忘模式 fire-and-forget 模式
producer.send(producerRecord)
- 异步发送发送
producer.send(producerRecord, new Callback() )
- 同步发送模式
producer.send(producerRecord, new Callback() ).get()
生产者核心参数
-
bootstrap.servers
: Kafka集群的地址列表,Producer用于发现Broker。这个参数是必需的。一般配置3-5个即可,如果你有100个broker,不需要全部配置,能连上一个就行 -
acks
: 默认值是all,控制生产者要求Broker确认消息写入的级别。要求消息可靠性设置-1
,要求吞吐量设置为1
。- 设置为0,表示生产端发送消息后立即返回,不等待broker端的响应结果。通常此时生产端吞吐量最高,消息发送的可靠性最低。
- 设置为1,表示leader副本成功写入PageCache就会响应Producer,而无需等待ISR(同步副本)集合中的其他副本写入成功。这种方案提供了适当的持久性,保证了一定的吞吐量。
- 设置成all或-1,表示不仅要等leader副本成功写入,还要求ISR中的其他副本成功写入,才会响应Producer。这种方案提供了最高的持久性,但也提供了最差的吞吐量。
-
retries
: 默认值Integer.MAX_VALUE,生产者在发送消息时的重试次数。如果消息发送失败,Producer会尝试重新发送。设置为大于0的值可以增加消息的可靠性,一般建议设置为3
。 -
retry.backoff.ms
: 两次重试之间的时间间隔,默认是 100ms。 -
max.in.flight.requests.per.connection
: 默认值5,建议 1-5,控制每个连接上允许的未确认请求的最大数量。较高的值可以提高吞吐量,但也可能导致更大的内存使用。 -
buffer.memory:
设置发送消息的缓冲区,默认是32M -
compression.type
: 默认是none,不压缩,但是也可以使用lz4压缩,效率还是不错的,但是会提高cpu的开销 -
batch.size
: 设置batch的大小,如果batch太小,会导致频繁的网络请求,吞吐量下降,如果batch太大,会导致一条消息需要等待很久才能被发送出去,而且会让内存缓冲区有很大的压力过多的数据缓存在内存里,默认值是16KB,也就是一个batch满了的时候会把这个batch发送出去,一般在生产环境会适当的增大这个值,如果消息大小大于这个,那么就会使用消息大小(源码中是这样的),一般不改动默认值 -
linger.ms
: 这个值默认是0,意思就是消息必须立即被发送,但是这样是不对的,一般设置一个100ms,这样的话就是如果100ms内这个batch满了16KB就会发送出去,如果是0 那么 batch.size 就失效,建议要求实时性设置0
,要求吞吐量设置50-100
-
key.serializer
和value.serializer
: 指定消息的键和值的序列化器,将Java对象转换为字节数组以便发送到Kafka。 -
enable.idempotence
:默认true,用于确保生产者产生的消息具有幂等性。幂等性是指无论发送多少次相同的消息,最终的结果都是相同的,不会产生副作用,仅能保证单分区,仅该生成者产生的消息幂等,重启下就废了。 -
max.block.ms
: 默认60s,控制在生产者缓冲区已满时调用send方法的行为。如果生产者缓冲区已满,send方法可以选择阻塞等待空间可用,直到指定的超时时间过去。 -
block.on.buffer.full
: 默认true,TRUE表示当我们内存用尽时,停止接收新消息记录或者抛出错误。
默认情况下,这个设置为TRUE。然而某些阻塞可能不值得期待,因此立即抛出错误更好。如果设置为false,则producer抛出一个异常错误:BufferExhaustedException
-
request.timeout.ms
: 默认30s,设置一个请求最大等待时间(单位为ms),超过这个时间则会抛Timeout异常。
超时时间如果设置大一些,如127000(127秒),高并发的场景中,能减少发送失败的情况。
以上Kafka版本为3.X
实际上,当batch.size和linger.ms二者都配置的时候,只要满足其中一个要求,就会发送请求到broker上。文章来源:https://www.toymoban.com/news/detail-697111.html
参考文章来源地址https://www.toymoban.com/news/detail-697111.html
- https://www.clairvoyant.ai/blog/unleash-kafka-producers-architecture-and-internal-workings
- 尚硅谷 Kafka
到了这里,关于图解系列 图解Kafka之Producer的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!