kafka 之 producer篇

前面三章从 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的那一套。