kafka

---

Kafka Producer将用户的消息发送到Kafka集群（准确讲是发送到Broker）。本文将分析Producer相关的代码实现。

类kafka.producer.Producer

如果你自己实现Kafka客户端来发送消息的话，你就是用到这个类提供的接口来发送消息。（如果你对如何利用Producer API来发送消息还不是很熟悉的话，可以参看官方的例子）。这个类提供了同步和异步两种方式来发送消息。

异步发送消息是基于同步发送消息的接口来实现的。异步发送消息的实现很简单，客户端消息发送过来以后，先放入到一个队列中然后就返回了。Producer再开启一个线程（ProducerSendThread）不断从队列中取出消息，然后调用同步发送消息的接口将消息发送给Broker。

Producer发送同步消息是委托给EventHandler做的，EventHandler是个接口，具体实现为DefaultEventHandler。它们的简化类图如下： 

可以看到，Producer类中的成员producerSendThread和queue是为了发送异步消息的，eventHandler是为了发送同步消息的，当然异步消息也需要它。KeyedMessage是封装了用户发送的消息。Seq是Scala中的序列，可以看成是Java中的List。

KeyedMessage的简化类图如下： 

类DefaultEventHandler

DefaultEventHandler是接口EventHandler唯一的实现。从上一节可以看到Producer发送给EventHandler的消息格式为KeyedMessage。我们来看看在将KeyedMessage发送给Broker之前需要做哪些工作。

序列化

KeyedMessage中的KV是由用户指定的自定义类型，而在发送给broker的时候是以二进制流来发送，因此还需要将用户自定类型的数据转换为二进制流。在初始化Producer的时候需要配置serializer.class，它就是用来处理这个事情的。此外，还要把多条Message组合成MessageSet并按照用户指定的压缩方式进行压缩。

找到对应的broker

KeyedMessage中指定了该Message的topic，而一个topic可以有多个partition，每个partition有多个replica，由多个Broker来管理，这些Broker中有一个leader。只有leader broker能够响应客户端的读写请求。

由此可见，在将KeyedMessage发送给broker之前，必须找到该条Message对应的leader broker，具体步骤为： 
1. 找出该topic的所有partition， 
2. 找出该KeyedMessage应该发送到的那个partition，在初始化Producer的时候配置partitioner.class就是用来对Message进行分区的 
3. 找出对应partition所在的leader broker。

最后，DefaultEventHandler将序列化后的Message封装成ProducerRequest，它自身并没有将ProducerRequest发送给broker的逻辑，而是将其交给SyncProducer来继续后面发送的流程。

ProducerPool, SyncProducer和BlockingChannel

它们在一起是完成最后的数据发送任务。先来看它们的类图： 
 
ProducerPool中有一个HashMap，其key为brokerid，value为连接到这个broker的SyncProducer。因此ProducerPool的更准确名字应该为SyncProducerPool。

BlockingChannel可以看成是一个Socket客户端，它有两个成员变量分别是机器名和端口号。它的connect方法会打开到对应机器的socket。它的send方法可以发送RequestOrResponse，它是真正发送数据的地方。

SyncProducer提供了两个send方法，分别用来发送ProducerRequest和TopicMetadataRequest。它内部是调用了blockingChannel来发送数据的。

小结

Producer发送数据的简化序列图如下：
从图中可见，各个类的职责明确，BlockingChannel负责最底层的数据发送，SyncProducer负责将Request发送到一个指定的Broker那里，DefaultEventHandler负责数据转换和选择正确的Broker，直接给客户端使用的Producer则在此基础上提供了同步和异步两种发送方式。