1. 概述

一、Kafka Producer包含哪些部分 其实我们讨论producer时，指的是用户接触的clients.producer这个util包，其中包含了发送数据到哪台机器、怎样序列化、分批发送与拒绝消息等等的发送策略。 但这些讨论不包括Kafka接到这些消息后怎样处理的问题，这些是broker（即kafka server端）需要去处理的部分。 二、Producer与Broker的关系 producer会维护将要发送的topic表，在必要时向broker要这些topic的元信息，并在client端维护这些元信息，来决定每条消息的去向。 从broker角度来看，它提供了元信息给producer以后，就无法控制producer的行为了。可以这么理解——broker本身是无状态的机器，主要提供的是元信息的消息读写的接口。 三、Producer发送消息的流程简介

在发送消息时，等待元信息的更新将key\value序列化为byte[]，计算出数据的大小是否超出限制计算出key对应的partition，以确认将消息发往哪里使用accumulator将数据放到topic\partition对应的缓冲区中使用NetworkClient类（背后是nio selector）定期的把缓冲区中的内容发出

注：本文基于Kafka 0.9版本，KafkaProducer的代码更新时间在2015-12-06

2. 字段与主要方法

2.1 KafkaProducer中的字段

private static final AtomicInteger PRODUCER_CLIENT_ID_SEQUENCE = new AtomicInteger(1); private String clientId; // client.id参数：用于日志与打点，以及发请求带的参数 private final Partitioner partitioner; // partitioner.class参数：指定一个Partitioner类用于计算key对应的分片，可自定义 private final int maxRequestSize; // max.request.size参数：单条消息的内存byte数大于此值时，抛出RecordTooLargeException异常 private final long totalMemorySize; // buffer.memory参数：同上，但额外作为accumulator->BufferPool中的`The maximum amount of memory that this buffer pool can allocate`，似乎是accumulator中暂存消息最多可用的内存量。 private final Metadata metadata; // 维护了一些topic的元信息 private final RecordAccumulator accumulator; // 用于堆积消息，按情况batch发送 private final Sender sender; // 发送消息 private final Thread ioThread; // 把this.sender包装成一个Thread private final CompressionType compressionType; //compression.type参数，压缩数据的方式，主要用在accumulator内。有none\gzip\snappy\lz4几种 private final Metrics metrics; // TODO 监控信息？ private final Sensor errors; // TODO 监控信息？ private final Time time; // KafkaProducer对象的创建时间 private final Serializer<K> keySerializer; // key序列化方法 private final Serializer<V> valueSerializer; // value序列化方法 private final ProducerConfig producerConfig; // 外部把配置传入，但没多大用处 private final long maxBlockTimeMs; // 来源不明TODO，用处似乎是限制send的一些IO操作总时间（waitOnMetadata堵塞的时间，与accumulator.append中waitMemory堵塞的时间之和，要低于此值） private final int requestTimeoutMs; //参数`timeout.ms` or `request.timeout.ms`，用途不明TODO

2.2 KafkaProducer提供的api

2.2.1 发送一条Kafka消息

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) 该方法做一定预处理后（详见第3节），将消息塞进accumulator缓冲区后，等待异步发送并返回，在消息实际发送完，即取到请求的返回值后，执行callback参数的回调方法。返回体中包装的RecordMetadata类型，带有消息最终发到的partition与offset，可用future.get()堵塞到消息发送完成。官方建议：由于callback会执行一个IO线程，因此建议处理速度足够快，否则会影响到其它线程。如果需要执行堵塞或计算量大的逻辑，建议另开一个Executor来并发处理callback来的数据。可能抛出的异常：InterruptException（堵塞时被打断）、SerializationException（序列化失败）、BufferExhaustedException（buffer空间用尽）

2.2.2 刷新缓存

public void flush()

此方法会将accumulator缓冲区中，尚未发出的那些消息变成可发送状态。调用此方法会让当前线程堵塞住，并保证调用flush()之前曾发送的所有消息都执行完成，即kafka服务端确实收到这些消息。 值得注意的是，在调用flush陷入堵塞时，其它线程仍然可以继续调用send发送消息。 以下是官方举例的一个使用case：

// 消费100条消息，打入另一个topic后，确保所有数据发送成功，再commit消费进度 for(ConsumerRecord<String, String> record: consumer.poll(100)) producer.send(new ProducerRecord("my-topic", record.key(), record.value()); producer.flush(); consumer.commit();

2.2.3 topic的partition信息

public List<PartitionInfo> partitionsFor(String topic)

可获取指定topic的partition元信息，若本地metadata缓存中没有此topic，则堵塞等待更新元信息。 超时时间定为maxBlockTimeMs，堵塞可被打断抛出InterruptException。

2.2.4 指标信息

public Map<MetricName, ? extends Metric> metrics()

获取监控信息，或说kafka使用过程中的打点。（TODO待补充）

2.2.5 关闭producer

public void close(long timeout, TimeUnit timeUnit)

关闭当前producer，堵塞当前线程等待所有sent请求完成。

3. 发送细节与机制详述

3.1 获取Metadata

以下为Metadata中的字段

private final long refreshBackoffMs; // 更新失败时最小的再次刷新间隔时间 private final long metadataExpireMs; // 过期时间, 默认60s private int version; // 每次更新version自增 private long lastRefreshMs; // 最近的更新时的时间 private long lastSuccessfulRefreshMs; // 最近成功更新的时间 private Cluster cluster; // 保存topic与partition、结点与partition等等的关系 private boolean needUpdate; // 需要更新 metadata private final Map<String, Long> topics; // topic与对应的过期时间的对应关系 private final List<Listener> listeners; // 事件监控者 private boolean needMetadataForAllTopics; // 是否强制更新所有的 metadata

producer在发送前执行waitOnMetadata(String topic, long maxWaitMs)来获取元信息。此方法判断缓存中是否有topic信息，没有的话进入以下循环：

int version = metadata.requestUpdate(); // 告诉metadata信息需要更新了 sender.wakeup(); // 唤醒sender来更新metadata metadata.awaitUpdate(version, remainingWaitMs); // 进入堵塞直到版本更新

上述的awaitUpdate操作中，metadata会堵塞住，等待sender的一系列请求成功后调用Metadata.update来唤醒自己。sender是kafka client的一个network util包装，后面再详细介绍。

如果这个过程超时了，或者topic unauthorized（TODO：想想unauthorized是什么情况？），就开始抛出异常。如果没有异常的顺利完成，topic对应的partition信息就会被producer收到并cache在内存中，producer可用此信息来发送消息。

3.2 消息的发送与缓冲区

一、计算partition与结点 经过3.1中讨论的元信息获取，producer就获取了这条消息topic有几个partition（分片）以及这些分片对应的结点信息。 需要发往哪一个partition：

观察消息是否指定了partition，未指定的话走partition类计算数据分片；默认partitioner类：如果消息带有key则按hash(keyBytes)%numPartitions，不带有key则按round-robin轮流发往各个partition可以通过partitioner.class配置，自定义partitioner类，按照业务需求来自定义partition计算方式。 确认了消息发往哪一个partition，也就能从元信息中找到partition对应的leader结点了。

二、accumulator提供的消息缓冲 确认了消息要发送的结点后，下一步就是将消息内容与结点信息塞进accumulator。 accumulator是一个buffer，负责把producer发送的消息累积在那里，直到特定条件时发出。其buffer采用一个map<TopicPartition, Deque >来存储，每一个topic\partition下都是一个RecordBatch队列。 队列中每一个RecordBatch元素，都代表一组kafka消息。在一个RecordBatch满了，或者其它线程调用flush()\close()等操作时，producer都会唤醒sender，来负责把buffer中的RecordBatch发到kafka服务端去。

3.3 消息缓冲的flush机制

在accumulator中使用了一个AtomicInteger flushesInProgress来表示是否在flush状态。 在Sender类（在4.2中会详细介绍Sender类）每次迭代时，会先算出现在要往哪些结点发送消息。计算逻辑是一个缓冲区满了、超时、没空间，或当前在flush状态，就会认为消息需要发出去了。 因此：当我们将flushesInProgress加一，则会在Sender下一轮迭代时，将所有待发数据的结点全标成待发送的。flush()流程示意如下：

将flushesInProgress++将sender唤醒去干活等待所有未完成的缓冲消息发送完成将flushesInProgress–

因此我们可以理解为，flush()操作会将当前所有缓存都发出一次。这就保证了在flush()前所有消息都会被发送完成，flush状态才结束；但保持flush状态的过程中新加入的消息，我们无法确定其状态。

4. 拓展：Sender

前面提到更新metadata，或发送accumulator中的批量消息，以及flush()，此类网络I/O操作，都会执行一个相同的代码sender.wakeup()。所以这个万能的sender到底是什么东西呢？

4.1 初始化与启动

下面是producer的构造器中，把Sender初始化出来的地方：

this.sender = new Sender(client, this.metadata, this.accumulator, config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG), (short) parseAcks(config.getString(ProducerConfig.ACKS_CONFIG)), config.getInt(ProducerConfig.RETRIES_CONFIG), this.metrics, new SystemTime(), clientId, this.requestTimeoutMs); String ioThreadName = "kafka-producer-network-thread" + (clientId.length() > 0 ? " | " + clientId : ""); this.ioThread = new KafkaThread(ioThreadName, this.sender, true); this.ioThread.start();

producer构造了一个Runnable的sender的成员变量，并开启线程执行。

4.2 Sender做了什么

其主要工作是不断的循环以下操作：

通过accumulator中可发出的数据包，以及metadata信息，算出现在准备发送消息的结点。（若找不到leader信息，标记一下更新metadata）"倒出"accumulator中的数据（把超时未发的数据打个记录），把数据包和对应结点打包创建出ClientRequest。把这些请求塞进sender中维护的client（使用nio维护的请求队列），并通过client.poll来发送出去。（TODO 网络相关的操作补充）poll()的时候会执行一个metadataUpdater.maybeUpdate(now)的操作： 先计算出下次刷新时间：根据是否请求刷新元数据（见本节第1条及3.1节中的刷新策略），或metadata.max.age.ms配置中的最久没有成功刷新的时间，并考虑到retry.backoff.ms配置中的最短刷新间隔，得出一个下次刷新的时间。考虑timeToNextReconnectAttempt，以及是否正在获取metadata信息，来更新下次刷新时间。如果已经到了可刷新时间点，获取leastLoadedNode，并将打向这个node的请求塞进client，也就搭上了这一波client.poll网络请求的车。

可以举个例子帮助解释：

第一轮迭代中，如果需要某些node A的元信息并发现缺失了，此时sender会请求更新metadata。那么在当前这一轮迭代结束的poll(now)操作中，通过metadataUpdater.mayBeUpdate(now)发现需要更新metadata，并发出更新的网络请求。第二轮迭代中：由于metadata的更新请求只发出没处理，因此依然没有缺失node的信息。但是在这一轮迭代结束的poll()中，如果metadata请求返回了，就可以通过返回值补上node A的信息。第三轮迭代中：可正常获取node A的信息，并向A发消息了。

5. 其它

TODO，待补充kafka 0.9到现在的更新内容