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的那一套。
页:
[1]