Kafka 0.10 Producer网络流程简述

1.Producer 网络请求

1.1 Producer Client角度

KafkaProducer主要靠Sender来发送数据给Broker。

Sender: 该线程handles the sending of produce requests to the Kafka cluster. 该线程发送metadata requests来更新它感知的整个集群的视图;另外，主要负责发送produce请求到相关的broker。
Selector的主要目的是网络事件的 loop 循环，通过调用selector.poll，不断轮询每个Channel上读写事件。
SocketChannel注册到Selector，Selector轮询到事件之后，让SocketChanel和ServerSocketChannel进行通信，做实际的IO等操作。

关注三个方法

1.注册事件(connect,accept,read,write)
2.轮询IO是否就绪
3.执行实际IO等操作。

思考一下，connect, accept, read, write 这4种事件，分别在这3个阶段对应的函数。

Producer发送请求的调用顺序：
KafkaProducer -- Sender -- KafkaClient（NetworkClient） -- Selector -- KafkaChannel

1.2 一次请求的详细过程

KafkaProducer 构造函数启动了Sender线程
Sender.run 调用 NetworkClient的send()函数，调用了selector.send(Send send)，但这个时候数据并没有真的发送出去，只是暂存在了selector内部相对应的KafkaChannel里面。

KafkaChannel先进行了检查，是否存在send的目的地，这是一个2轮询。确保有相应的KafkaChannel之后，调用this.transportLayer.addInterestOps(SelectionKey.OP_WRITE); 1注册 write事件。接下来就交给Selector进行2轮询 和3实际操作，详细方法是Selector.poll。

KafkaProducer.send 调用doSend方法，首先调用waitOnMetadata获取metaData信息，最后调用的是nioSelector.wakeUp(),让阻塞在select()的Selector立即返回，准备IO事件。（在send之前，会先读取metadata。如果metadata读不到，会一直阻塞在那，直到超时，抛出TimeoutException）

1.3 `Selector`处理注册的事件

小知识

一个Selector可以处理多个Channel。
SelectionKey用来记录一个Channel上的事件集合，每个Channel对应一个SelectionKey。
SelectionKey也是Selector和Channel之间的关联，通过SelectionKey可以取到对应的Selector和Channel。

`poll 和 pollSelectionKeys` 的关键流程（正常情况的处理流程）

我们以write事件coming来举例，当有事件到来的时候，

找到该事件对应的SocketChannel（即KafkaChannel），为了后续与对应的ServerSocketChannel进行通信

如果这个KafkaChannel是可用的，在channel不是ready的状态下，会channel.prepare进行初始化, 里面包括了权限认证。（会调用下面的这个类进行权限认证，这里也是出过问题的地方：SaslServerAuthenticator#handleKafkaRequest）

轮询事件的类型，connect, read, write

write事件的情况下，调用Send send = channel.write(); // write--阶段3: 实际的IO操作, 读取完数据后,就取消write事件.

期间出现过任何异常，都会关闭这个KafkaChanel（上面的授权都没有了），常见的是IOException异常,Server端日志经常出现。

KafkaChannel的授权创建是在Selector的connect、register方法中

2 同步和异步

Producer有同步发送和异步发送2种策略。在以前的Kafka client api实现中，同步和异步是分开实现的。

而在0.9以后的版本中，同步发送其实是通过异步发送间接实现，其接口如下：

public class KafkaProducer<K, V> implements Producer<K, V> {
...
    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback)  //异步发送接口
     {
     ...
     }
}

要实现同步发送，只要在拿到返回的Future对象之后，直接调用get()就可以了。

2.1 基本思路

异步发送的基本思路就是：send的时候，KafkaProducer把消息放到本地的消息队列RecordAccumulator，然后一个后台线程Sender不断循环，把消息发给Kafka集群。

要实现这个，还得有一个前提条件：就是KafkaProducer/Sender都需要获取集群的配置信息Metadata。

所谓Metadata：Topic/Partion与broker的映射关系：每一个Topic的每一个Partition，得知道其对应的broker列表是什么，其中leader是谁，follower是谁。