通过上一课时的介绍我们了解到，业务线程使用 KafkaProducer.send() 方法发送 message 的时候，会先将其写入RecordAccumulator 中进行缓冲，当 RecordAccumulator...本课时我们就重点来看一下 RecordAccumulator 这个缓冲区的结构。...首先，我们从上图中可以看出，RecordAccumulator 会由业务线程写入、Sender 线程读取，这是一个非常明显的生产者-消费者模式，所以我们需要保证 RecordAccumulator 是线程安全的...RecordAccumulator 、ProducerBatch、MemoryRecordsBuilder 这三个核心类的关系如下图所示： message 格式 既然我们准备深入 KafkaProducer...、ProducerBatch、BufferPool 等底层组件，以及 RecordAccumulator 的核心方法。
〇、前言 在上一篇文章《连Producer端的主线程模块运行原理都不清楚，就敢说自己精通Kafka》中，我们介绍了Main Thread的工作原理，那么在本篇文章中，我们继续介绍第二部分内容：RecordAccumulator...一、RecordAccumulator 在上文中，我们介绍了主线程（Main Thread）的执行流程，当我们使用KafkaProducer发送消息的时候，消息会经过拦截器（Interceptor）、序列化器...（Serializer）和分区器（Partitioner），最后会暂存到消息收集器（RecordAccumulator）中，那么，本节就来针对其进行介绍。...RecordAccumulator的主要作用是暂存Main Thread发送过来的消息，然后Sender Thread就可以从RecordAccumulator中批量的获取到消息，减少单个消息获取的请求次数...properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60*1000); 在RecordAccumulator中，我们通过getOrCreateDeque
①main线程中创建了一个双端队列RecordAccumulator，将消息发送给RecordAccumulator。...发送消息到RecordAccumulator：主线程将创建的消息发送到一个双端队列（deque）叫做RecordAccumulator。...Sender线程的主要工作如下： 从RecordAccumulator拉取消息：Sender线程定期轮询（poll）RecordAccumulator，检查是否有新消息需要发送。...RecordAccumulator： RecordAccumulator是Producer内部的一个共享变量，用于暂存即将发送到Kafka broker的消息。...发送原理小结 总结一下，Kafka的Producer采用异步发送消息的方式， 主线程负责创建和发送消息到RecordAccumulator， 而Sender线程负责从RecordAccumulator
主线程: 负责将消息发送到消息累加器(RecordAccumulator) . Sender线程: 负责将消息累加器(RecordAccumulator)中获取消息并发送到Broker....RecordAccumulator: 主要用来缓存消息, Sender 线程可以批量发送，进而减少网络传输的资源消耗以提升性能 RecordAccumulator 缓存的大小可以通过生产者客户端参数 buffer.memory...主线程中发送过来的消息都会被加到 RecordAccumulator 的某个双端队列（ Deque ）中， RecordAccumulator 的内部为每个分区都维护了 个双端队列，队列中的内容就是 Producer...在 RecordAccumulator 的内部还有一个 BufferPool, 它主要用来实现 ByteBuffer 的复用，以实现缓存的高效利用 。
Kafka 生产者结构 Kafka Producer 主要有三个部分组成：主线程、Sender 线程、RecordAccumulator。...主线程：执行序列化、分区、拦截器处理等主要操作，并将消息缓存到 RecordAccumulator 中； Sender 线程：从 RecordAccumulator 中拉数据，发送到 broker； RecordAccumulator...：缓存主线程的消息，并提供给 Sender 线程； 生产者执行流程： 主线程处理数据，包括处理拦截器操作、序列化、分区； 主线程将数据发送给 RecordAccumulator； RecordAccumulator...ProducerRecord) 压缩成 (ProducerBatch)，并以分区 (TopicPartition) - 消息队列 (Deque(ProducerBatch)) 的消息存储主线程发来的消息； Sender 从 RecordAccumulator
，主要用于存放消息（KafkaProducer主线程往RecordAccumulator中写消息，Sender线程从RecordAccumulator中读消息并发送到Kafka中） 6.解析Broker...// 唤醒Sender线程 Sender.wakeup RecordAccumulator RecordAccumulator是消息队列用于缓存消息，根据TopicPartition对消息分组 重点看下...RecordAccumulator.applend追加消息的流程 // 记录进行applend的线程数 appendsInProgress.incrementAndGet(); // 根据TopicPartition...// 根据准备好的节点信息从缓冲区中获取topicPartion对应的Deque队列中取出ProducerBatch信息 RecordAccumulator.drain // 将消息转移到每个节点的生产请求队列中...通过上面的介绍，我们梳理出了Kafka生产消息的主要流程，涉及到主线程往RecordAccumulator中写入消息，同时后台的Sender线程从RecordAccumulator中获取消息，使用NIO
在消息发送的过程中，涉及到了两个线程——main线程和Sender线程，以及一个线程共享变量——RecordAccumulator。...main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。..."batch.size", 16384);//批次大小 props.put("linger.ms", 1);//等待时间 props.put("buffer.memory", 33554432);//RecordAccumulator...16384);//批次大小 props.put("linger.ms", 1);//等待时间 props.put("buffer.memory", 33554432);//RecordAccumulator...16384);//批次大小 props.put("linger.ms", 1);//等待时间 props.put("buffer.memory", 33554432);//RecordAccumulator
在主线程中由KafkaProducer创建消息，然后通过可能的拦截器、序列化器和分区器的作用之后缓存到消息累加器（RecordAccumulator，也称为消息收集器）中。...Sender线程负责从RecordAccumulator中获取消息并将其发送到Kafka中。...RecordAccumulator主要用来缓存消息以便Sender线程可以批量发送，进而减少网络传输的资源消耗提升性能。...RecordAccumulator缓存的大小可以通过生产者客户端参数buffer.memory的配置，默认值为32MB。...主线程中发送过来的消息都会被追回到RecordAccumulator的某个双端队列（Deque）中，在RecordAccumulator的内部为每个分区都维护了一个双端队列，队列中的内容就是ProducerBatch
将 message 封装成 ProducerRecord 写入到 RecordAccumulator 中暂存，RecordAccumulator 对象中维护了多个队列，可以看做是 message 的缓冲区...调用 append()方法，将 message 写入到 RecordAccumulator 中暂存。...最后，唤醒 Sender 线程，后续就由 Sender 线程从 RecordAccumulator 中批量发送 message 到 kafka 集群。...我们前面介绍整个 KafkaProducer 流程的时候说过，RecordAccumulator 是一个缓冲区，主线程发送的 message 会先进入 RecordAccumulator，然后 Sender...我们看一下 KafkaProducer.doSend()方法中，有这么一个片段： // 尝试向RecordAccumulator中追加message RecordAccumulator.RecordAppendResult
RecordAccumulator 该类扮演的是一个队列的角色，将records追加到MemoryRecords实例中，用于发送到server端。 RecordBatch 一批准备发送的消息。...RecordAccumulator维护了一个ConcurrentMap> batches; RecordBatch维护了一个MemoryRecords...F),构建了一个RecordAccumulator。此时需要关注的两个配置是 batch.size：批量发送的大小。 linger.ms：超时发送的时间。...获取分区号 int partition = partition(record, serializedKey, serializedValue, metadata.fetch()); 3),将消息追加到RecordAccumulator...//将消息追加到RecordAccumulator RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp
Sender线程 负责从RecordAccumulator中获取消息并发送到Kafka中。...❞ 「RecordAccumulator」 ❝ RecordAccumulator主要用来缓存消息以便Sender线程可以批量发送，进而减少网络传输的资源消耗以提升性能。...RecordAccumulator缓存的大小可以通过生产者客户端参数buffer.memory进行配置，默认值是32MB。...❞ 「ProducerBatch」 ❝ Main线程发送过来的消息会被追加到RecordAccumulator的Deque(双端队列)中，在RecordAccumulator的内部每个Partition...❞ 源码分析及图解原理 RecordAccumulator 在RecordAccumulator中，最核心的参数就是： private final ConcurrentMap<TopicPartition
再往下看，到了第一个重点：RecordAccumulator self...._accumulator = RecordAccumulator(message_version=message_version, metrics=self._metrics, **self...._batches = collections.defaultdict(collections.deque) RecordAccumulator类是啥呢？...里 RecordAccumulator的大小由参数buffer_memory控制，batch的大小由参数batch_size控制。...send方法实际只是往RecordAccumulator容器里塞消息，Sender线程则在后台不停轮训，符合条件就发送。
调用RecordAccumulator.append()方法，将处理后的数据扔进RecordAccumulator（缓存对象)的RecordAppendResult类属性中； 5....RecordAccumulator.append()方法首先将数据进行队列化放在Deque对象中，Deque包含多个ProducerBatch； 6....对象的RecordAccumulator.append()方法，返回RecordAppendResult对象； RecordAccumulator.RecordAppendResult result =...，RecordAccumulator持有RecordAppendResult对象，其future作为整个producer.send()方法的返回值； public final static class...通过getOrCreateDeque(tp)得到deque队列（持有ProducerBatch对象），ProducerBatch是最小的发送数据实体，RecordAccumulator计算字节数并分配本地资源
2、然后调用上一课时介绍的 RecordAccumulator.ready() 方法，了解 RecordAccumulator 的缓存情况，选出可以向哪些Node 节点发送请求。...expiredBatches() 方法用来获取 RecordAccumulator 中已经过期的 ProducerBatch 集合。...Selector 通过上面的小节，我们已经了解了 RecordAccumulator 中的数据转换成 ClientRequest 的流程。...如果 RecordAccumulator 中有部分数据，但发送条件没有 ready，则 timeout 为消息的超时时长。...如果 RecordAccumulator 没有数据，则 timeout 为 kafka 元数据过期的时长。
在消息发送的过程中，涉及到了两个线程——main线程和Sender线程，以及一个线程共享变量——RecordAccumulator。...main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。...16384);//批次大小 props.put("linger.ms", 1);//等待时间 props.put("buffer.memory", 33554432);//RecordAccumulator...16384);//批次大小 props.put("linger.ms", 1);//等待时间 props.put("buffer.memory", 33554432);//RecordAccumulator
2、RecordAccumulator：是一个记录收集器，用于收集客户端发送的消息，并将收集到的消息暂存到客户端缓存中。...流程详解：消息发送的过程中，涉及到两个线程协同工作：1、主线程首先将业务数据封装成ProducerRecord对象，2、之后调用send()方法将消息放入RecordAccumulator(消息收集器，...也可以理解为主线程与Sender线程直接的缓冲区)中暂存，3、Sender线程负责将消息信息构成请求，并最终执行网络I/O的线程，它从RecordAccumulator中取出消息并批量发送出去， 需要注意的是...hash，然后对分区数取模(Utils.murmur2(key) % numPartitions) ~ 3、如未指定且没key，则轮询发送给分区（低版本采用随机）5、临时缓存(存储) RecordAccumulator...采用了双端队列(Deque)数据结构来临时存储 目的：提高发送数据的吞吐量 确定消息发送的分区后，会在RecordAccumulator寻找对应的Deque，找不到对应的Deque则新建
NetworkClient中的InFlightRequests、Sender中的inFlightBatches变量和RecordAccumulator中的IncompleteBatches，都用到了这个机制...incomplete 在RecordAccumulator::append中，内存被申请，该Batch被添加到incomplete。...在响应收到后，RecordAccumulator::deallocate被调用，移除对应的Batch，释放内存 ---- Sender::sendProduceRequest中定义的RequestCompletionHandler
RecordAccumulator accumulator 消息记录累积器，消息追加的入口(RecordAccumulator 的 append 方法)。...由于在讲解 Sender 发送流程中，大部分都是调用 RecordAccumulator 方法来实现其特定逻辑，故接下来重点对上述涉及到 RecordAccumulator 的方法进行一个详细剖析，加强对...2、RecordAccumulator 核心方法详解 2.1 RecordAccumulator 的 ready 方法详解 该方法主要就是根据缓存区中的消息，判断哪些分区已经达到发送条件。...2.2 RecordAccumulator 的 drain方法详解 RecordAccumulator#drain public Map> drain...RecordAccumulator#drainBatchesForOneNode private List drainBatchesForOneNode(Cluster cluster
kafka的java producer的发送是异步，主要的分几步： append到RecordAccumulator sender从RecordAccumulator取出RecordBatch，交给client...打交道，把RecordBatch发送出去 这里就涉及到了几个步骤的异常，append的时候，会抛异常，对于ApiException则放到callback里头去，其他异常直接抛出来(callback仅仅是跟RecordAccumulator...with callback {} to topic {} partition {}", record, callback, record.topic(), partition); RecordAccumulator.RecordAppendResult...Cluster cluster = metadata.fetch(); // get the list of partitions with data ready to send RecordAccumulator.ReadyCheckResult
领取专属 10元无门槛券
手把手带您无忧上云
云服务器
即时通信 IM
ICP备案
对象存储
实时音视频
