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 一次请求的详细过程

  1. KafkaProducer 构造函数启动了Sender线程
  2. 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

  1. 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来举例,当有事件到来的时候,

  1. 找到该事件对应的SocketChannel(即KafkaChannel),为了后续与对应的ServerSocketChannel进行通信

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

  1. 轮询事件的类型,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是谁。

上一篇:动态更新UI的方式


下一篇:TP3.2.3 接入支付宝