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是谁。