kafka 之 producer篇

yw6866

　　前面三章从 kafka.kafka 启动类 抽丝剥茧，看了 kafka broker 的架构。
　　下面咱们从 producer 的角度，producer的逻辑相对简单，再看看 client 端，和 broker 都干了什么。
　　首先，咱们看看官方给出的 examples， 在包中的examples\src 下有给出的示例，java版的，搞到eclipse中。

   producer = new kafka.javaapi.producer.Producer<Integer, String>(new ProducerConfig(props));

　　用了kafka中的 javaapi 中的那个包。
　　只调用了一个方法：

  public void run() {
int messageNo = 1;
while(true)
{
String messageStr = new String("Message_" + messageNo);
producer.send(new ProducerData<Integer, String>(topic, messageStr));
messageNo++;
}
}

　　点进去看看 send 方法，基本做了两件大事情， 从zk中获取broker信息， 发。

        brokerPartitionInfo = new ZKBrokerPartitionInfo(new ZKConfig(zkProps), producerCbk)

　　在 ZKBrokerPartitionInfo 中封装了一些对于zk的操作， 从zk中拿到 一个 topic 到 （brokerid，partition）的一个map，并且维护这个map。还有 brokerid -> (host, port) 的一个map。 这样就很容易拿到 要给哪个 broker发送信息。找到之后就是发了。

  def writeCompletely(channel: GatheringByteChannel): Int = {
var written = 0
while(!complete) {
written = writeTo(channel)
trace(written + " bytes written.")
}
written
}
　　写就简单了， nio的那一套。