Kafka消息发送线程及网络通信

也想出现在这里？点击联系我~

回顾一下前面提到的发送消息的时序图，上一节说到了Kafka相关的元数据信息以及消息的封装，消息封装完成之后就开始将消息发送出去，这个任务由Sender线程来实现。

1. Sender线程

找到KafkaProducer这个对象，KafkaProducer的构造函数中有这样几行代码。

1. this.accumulator=new
2. RecordAccumulator(config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
3. this.totalMemorySize,
4. this.compressionType,
5. config.getLong(ProducerConfig.LINGER_MS_CONFIG),
6. retryBackoffMs,
7. metrics,
8. time);

构造了RecordAccumulator对象，设置了该对象中每个消息批次的大小、缓冲区大小、压缩格式等等。

紧接着就构建了一个非常重要的组件NetworkClient，用作发送消息的载体。

1. NetworkClientclient=newNetworkClient(
2. newSelector(config.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),this.metrics,time,"producer",channelBuilder),
3. this.metadata,
4. clientId,
5. config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION),
6. config.getLong(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG),
7. config.getInt(ProducerConfig.SEND_BUFFER_CONFIG),
8. config.getInt(ProducerConfig.RECEIVE_BUFFER_CONFIG),
9. this.requestTimeoutMs,time);

对于构建的NetworkClient，有几个重要的参数要注意一下：

√ connections.max.idle.ms: 表示一个网络连接最多空闲多久，超过这个空闲时间，就关闭这个网络连接，默认是9分钟。

√ max.in.flight.requests.per.connection：表示每个网络连接可以容忍 producer端发送给broker 消息然后消息没有响应的个数，默认是5个。(ps：producer向broker发送数据的时候，其实是存在多个网络连接)

√ send.buffer.bytes：socket发送数据的缓冲区的大小，默认值是128K。

√ receive.buffer.bytes：socket接受数据的缓冲区的大小，默认值是32K。

构建好消息发送的网络通道直到启动Sender线程，用于发送消息。

1. this.sender=newSender(client,
2. this.metadata,
3. this.accumulator,
4. config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)==1,
5. config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
6. (short)parseAcks(config.getString(ProducerConfig.ACKS_CONFIG)),
7. config.getInt(ProducerConfig.RETRIES_CONFIG),
8. this.metrics,
9. newSystemTime(),
10. clientId,
11. this.requestTimeoutMs);
12. //默认的线程名前缀为kafka-producer-network-thread，其中clientId是生产者的id
13. StringioThreadName="kafka-producer-network-thread"+(clientId.length()>0?"|"+clientId:"");
14. //创建了一个守护线程，将Sender对象传进去。
15. this.ioThread=newKafkaThread(ioThreadName,this.sender,true);
16. //启动线程
17. this.ioThread.start();

看到这里就非常明确了，既然是线程，那么肯定有run()方法，我们重点关注该方法中的实现逻辑，在这里要补充一点值得我们借鉴的线程使用模式，可以看到在创建sender线程后，并没有立即启动sender线程，而且新创建了KafkaThread线程，将sender对象给传进去了，然后再启动KafkaThread线程，相信有不少小伙伴会有疑惑，我们进去KafkaThread这个类看一下其中的内容。

1. /**
2. *AwrapperforThreadthatsetsthingsupnicely
3. */
4. publicclassKafkaThreadextendsThread{
5. privatefinalLoggerlog=LoggerFactory.getLogger(getClass());
6. publicKafkaThread(finalStringname,Runnablerunnable,booleandaemon){
7. super(runnable,name);
8. //设置为后台守护线程
9. setDaemon(daemon);
10. setUncaughtExceptionHandler(newThread.UncaughtExceptionHandler(){
11. publicvoiduncaughtException(Threadt,Throwablee){
12. log.error("Uncaughtexceptionin"+name+":",e);
13. }
14. });
15. }
16. }

发现KafkaThread线程其实只是启动了一个守护线程，那么这样做的好处是什么呢?答案是可以将业务代码和线程本身解耦，复杂的业务逻辑可以在KafkaThread这样的线程中去实现，这样在代码层面上Sender线程就非常的简洁，可读性也比较高。

先看一下Sender这个对象的构造。

/**

* 主要的功能就是用处理向 Kafka 集群发送生产请求的后台线程，更新元数据信息，以及将消息发送到合适的节点

* The background thread that handles the sending of produce requests to the Kafka cluster. This thread makes metadata

* requests to renew its view of the cluster and then sends produce requests to the appropriate nodes.

1. publicclassSenderimplementsRunnable{
2. privatestaticfinalLoggerlog=LoggerFactory.getLogger(Sender.class);
3. //kafka网络通信客户端，主要用于与broker的网络通信
4. privatefinalKafkaClientclient;
5. //消息累加器，包含了批量的消息记录
6. privatefinalRecordAccumulatoraccumulator;
7. //客户端元数据信息
8. privatefinalMetadatametadata;
9. /*theflagindicatingwhethertheproducershouldguaranteethemessageorderonthebrokerornot.*/
10. //保证消息的顺序性的标记
11. privatefinalbooleanguaranteeMessageOrder;
12. /*themaximumrequestsizetoattempttosendtotheserver*/
13. //对应的配置是max.request.size，代表调用send()方法发送的最大请求大小
14. privatefinalintmaxRequestSize;
15. /*thenumberofacknowledgementstorequestfromtheserver*/
16. //用于保证消息发送状态，分别有-1，0，1三种选项
17. privatefinalshortacks;
18. /*thenumberoftimestoretryafailedrequestbeforegivingup*/
19. //请求失败重试的次数
20. privatefinalintretries;
21. /*theclockinstanceusedforgettingthetime*/
22. //时间工具，计算时间，没有特殊含义
23. privatefinalTimetime;
24. /*truewhilethesenderthreadisstillrunning*/
25. //表示线程状态，true则表示running
26. privatevolatilebooleanrunning;
27. /*truewhenthecallerwantstoignoreallunsent/inflightmessagesandforceclose.*/
28. //强制关闭消息发送的标识，一旦设置为true，则不管消息有没有发送成功都会忽略
29. privatevolatilebooleanforceClose;
30. /*metrics*/
31. //发送指标收集
32. privatefinalSenderMetricssensors;
33. /*paramclientIdoftheclient*/
34. //生产者客户端id
35. privateStringclientId;
36. /*themaxtimetowaitfortheservertorespondtotherequest*/
37. //请求超时时间
38. privatefinalintrequestTimeout;
39. //构造器
40. publicSender(KafkaClientclient,
41. Metadatametadata,
42. RecordAccumulatoraccumulator,
43. booleanguaranteeMessageOrder,
44. intmaxRequestSize,
45. shortacks,
46. intretries,
47. Metricsmetrics,
48. Timetime,
49. StringclientId,
50. intrequestTimeout){
51. this.client=client;
52. this.accumulator=accumulator;
53. this.metadata=metadata;
54. this.guaranteeMessageOrder=guaranteeMessageOrder;
55. this.maxRequestSize=maxRequestSize;
56. this.running=true;
57. this.acks=acks;
58. this.retries=retries;
59. this.time=time;
60. this.clientId=clientId;
61. this.sensors=newSenderMetrics(metrics);
62. this.requestTimeout=requestTimeout;
63. }
64. ….
65. }

大概了解完Sender对象的初始化参数之后，开始步入正题，找到Sender对象中的run()方法。

1. publicvoidrun(){
2. log.debug("StartingKafkaproducerI/Othread.");
3. //sender线程启动起来了以后就是处于一直运行的状态
4. while(running){
5. try{
6. //核心代码
7. run(time.milliseconds());
8. }catch(Exceptione){
9. log.error("UncaughterrorinkafkaproducerI/Othread:",e);
10. }
11. }
12. log.debug("BeginningshutdownofKafkaproducerI/Othread,sendingremainingrecords.");
13. //okaywestoppedacceptingrequestsbuttheremaystillbe
14. //requestsintheaccumulatororwaitingforacknowledgment,
15. //waituntilthesearecompleted.
16. while(!forceClose&&(this.accumulator.hasUnsent()||this.client.inFlightRequestCount()>0)){
17. try{
18. run(time.milliseconds());
19. }catch(Exceptione){
20. log.error("UncaughterrorinkafkaproducerI/Othread:",e);
21. }
22. }
23. if(forceClose){
24. //Weneedtofailalltheincompletebatchesandwakeupthethreadswaitingon
25. //thefutures.
26. this.accumulator.abortIncompleteBatches();
27. }
28. try{
29. this.client.close();
30. }catch(Exceptione){
31. log.error("Failedtoclosenetworkclient",e);
32. }
33. log.debug("ShutdownofKafkaproducerI/Othreadhascompleted.");
34. }

以上的run()方法中，出现了两个while判断，本意都是为了保持线程的不间断运行，将消息发送到broker，两处都调用了另外的一个带时间参数的run(xx)重载方法，第一个run(ts)方法是为了将消息缓存区中的消息发送给broker，第二个run(ts)方法会先判断线程是否强制关闭，如果没有强制关闭，则会将消息缓存区中未发送出去的消息发送完毕，然后才退出线程。

/**

* Run a single iteration of sending

*

* @param now

* The current POSIX time in milliseconds

*/

1. voidrun(longnow){
3. //第一步，获取元数据
4. Clustercluster=metadata.fetch();
5. //getthelistofpartitionswithdatareadytosend
7. //第二步，判断哪些partition满足发送条件
8. RecordAccumulator.ReadyCheckResultresult=this.accumulator.ready(cluster,now);
9. /**
10. *第三步，标识还没有拉取到元数据的topic
11. */
12. if(!result.unknownLeaderTopics.isEmpty()){
13. //Thesetoftopicswithunknownleadercontainstopicswithleaderelectionpendingaswellas
14. //topicswhichmayhaveexpired.Addthetopicagaintometadatatoensureitisincluded
15. //andrequestmetadataupdate,sincetherearemessagestosendtothetopic.
16. for(Stringtopic:result.unknownLeaderTopics)
17. this.metadata.add(topic);
18. this.metadata.requestUpdate();
19. }
20. //removeanynodeswearen'treadytosendto
21. Iterator<Node>iter=result.readyNodes.iterator();
22. longnotReadyTimeout=Long.MAX_VALUE;
23. while(iter.hasNext()){
24. Nodenode=iter.next();
25. /**
26. *第四步，检查与要发送数据的主机的网络是否已经建立好。
27. */
28. //如果返回的是false
29. if(!this.client.ready(node,now)){
30. //移除result里面要发送消息的主机。
31. //所以我们会看到这儿所有的主机都会被移除
32. iter.remove();
33. notReadyTimeout=Math.min(notReadyTimeout,this.client.connectionDelay(node,now));
34. }
35. }

/**

* 第五步，有可能我们要发送的partition有很多个，这种情况下，有可能会存在这样的情况

* 部分partition的leader partition分布在同一台服务器上面。

*

*

*/

1. Map<Integer,List<RecordBatch>>batches=this.accumulator.drain(cluster,
2. result.readyNodes,
3. this.maxRequestSize,
4. now);
5. if(guaranteeMessageOrder){
6. //Muteallthepartitionsdrained
7. //如果batches空的话，跳过不执行。
8. for(List<RecordBatch>batchList:batches.values()){
9. for(RecordBatchbatch:batchList)
10. this.accumulator.mutePartition(batch.topicPartition);
11. }
12. }

/**

* 第六步，处理超时的批次

*

*/

1. List<RecordBatch>expiredBatches=this.accumulator.abortExpiredBatches(this.requestTimeout,now);
2. //updatesensors
3. for(RecordBatchexpiredBatch:expiredBatches)
4. this.sensors.recordErrors(expiredBatch.topicPartition.topic(),expiredBatch.recordCount);
5. sensors.updateProduceRequestMetrics(batches);
6. /**

* 第七步，创建发送消息的请求，以批的形式发送，可以减少网络传输成本，提高吞吐

*/

1. List<ClientRequest>requests=createProduceRequests(batches,now);
2. //Ifwehaveanynodesthatarereadytosend+havesendabledata,pollwith0timeoutsothiscanimmediately
3. //loopandtrysendingmoredata.Otherwise,thetimeoutisdeterminedbynodesthathavepartitionswithdata
4. //thatisn'tyetsendable(e.g.lingering,backingoff).Notethatthisspecificallydoesnotincludenodes
5. //withsendabledatathataren'treadytosendsincetheywouldcausebusylooping.
6. longpollTimeout=Math.min(result.nextReadyCheckDelayMs,notReadyTimeout);
7. if(result.readyNodes.size()>0){
8. log.trace("Nodeswithdatareadytosend:{}",result.readyNodes);
9. log.trace("Created{}producerequests:{}",requests.size(),requests);
10. pollTimeout=0;
11. }
12. //发送请求的操作
13. for(ClientRequestrequest:requests)
14. //绑定op_write
15. client.send(request,now);
17. //ifsomepartitionsarealreadyreadytobesent,theselecttimewouldbe0;
18. //otherwiseifsomepartitionalreadyhassomedataaccumulatedbutnotreadyyet,
19. //theselecttimewillbethetimedifferencebetweennowanditslingerexpirytime;
20. //otherwisetheselecttimewillbethetimedifferencebetweennowandthemetadataexpirytime;
21. /**

* 第八步，真正执行网络操作的都是这个NetWordClient这个组件

* 包括：发送请求，接受响应(处理响应)

1. this.client.poll(pollTimeout,now);

以上的run(long)方法执行过程总结为下面几个步骤：

1. 获取集群元数据信息

2. 调用RecordAccumulator的ready()方法，判断当前时间戳哪些partition是可以进行发送，以及获取partition 的leader partition的元数据信息，得知哪些节点是可以接收消息的

3. 标记还没有拉到元数据的topic，如果缓存中存在标识为unknownLeaderTopics的topic信息，则将这些topic添加到metadata中，然后调用metadata的requestUpdate()方法，请求更新元数据

4. 将不需要接收消息的节点从按步骤而返回的结果中删除，只对准备接收消息的节点readyNode进行遍历，检查与要发送的节点的网络是否已经建立好，不符合发送条件的节点都会从readyNode中移除掉

5. 针对以上建立好网络连接的节点集合，调用RecordAccumulator的drain()方法，得到等待发送的消息批次集合

6. 处理超时发送的消息，调用RecordAccumulator的addExpiredBatches()方法，循环遍历RecordBatch，判断其中的消息是否超时，如果超时则从队列中移除，释放资源空间

7. 创建发送消息的请求，调用createProducerRequest方法，将消息批次封装成ClientRequest对象，因为批次通常是多个的，所以返回一个List集合

8. 调用NetworkClient的send()方法，绑定KafkaChannel的op_write操作

9. 调用NetworkClient的poll()方法拉取元数据信息，建立连接，执行网络请求，接收响应，完成消息发送

以上就是Sender线程对消息以及集群元数据所发生的核心过程。其中就涉及到了另外一个核心组件NetworkClient。

2. NetworkClient

NetworkClient是消息发送的介质，不管是生产者发送消息，还是消费者接收消息，都需要依赖于NetworkClient建立网络连接。同样的，我们先了解NetworkClient的组成部分，主要涉及NIO的一些知识，有兴趣的童鞋可以看看NIO的原理和组成。

1. /**
2. *Anetworkclientforasynchronousrequest/responsenetworki/o.Thisisaninternalclassusedtoimplementthe
3. *user-facingproducerandconsumerclients.
4. *<p>
5. *Thisclassisnotthread-safe!
6. */
7. publicclassNetworkClientimplementsKafkaClient
8. {
10. privatestaticfinalLoggerlog=LoggerFactory.getLogger(NetworkClient.class);
11. /*theselectorusedtoperformnetworki/o*/
12. //javaNIOSelector
13. privatefinalSelectableselector;
14. privatefinalMetadataUpdatermetadataUpdater;
15. privatefinalRandomrandOffset;
16. /*thestateofeachnode'sconnection*/
17. privatefinalClusterConnectionStatesconnectionStates;
18. /*thesetofrequestscurrentlybeingsentorawaitingaresponse*/
19. privatefinalInFlightRequestsinFlightRequests;
20. /*thesocketsendbuffersizeinbytes*/
21. privatefinalintsocketSendBuffer;
22. /*thesocketreceivesizebufferinbytes*/
23. privatefinalintsocketReceiveBuffer;
24. /*theclientidusedtoidentifythisclientinrequeststotheserver*/
25. privatefinalStringclientId;
26. /*thecurrentcorrelationidtousewhensendingrequeststoservers*/
27. privateintcorrelation;
28. /*maxtimeinmsfortheproducertowaitforacknowledgementfromserver*/
29. privatefinalintrequestTimeoutMs;
30. privatefinalTimetime;
31. ……
32. }

可以看到NetworkClient实现了KafkaClient接口，包括了几个核心类Selectable、MetadataUpdater、ClusterConnectionStates、InFlightRequests。

2.1 Selectable

其中Selectable是实现异步非阻塞网络IO的接口，通过类的注释可以知道Selectable可以使用单个线程来管理多个网络连接，包括读、写、连接等操作，这个和NIO是一致的。

我们先看看Selectable的实现类Selector，是org.apache.kafka.common.network包下的，源码内容比较多，挑相对比较重要的看。

1. publicclassSelectorimplementsSelectable{
2. publicstaticfinallongNO_IDLE_TIMEOUT_MS=-1;
3. privatestaticfinalLoggerlog=LoggerFactory.getLogger(Selector.class);
4. //这个对象就是javaNIO里面的Selector
5. //Selector是负责网络的建立，发送网络请求，处理实际的网络IO。
6. //可以算是最核心的一个组件。
7. privatefinaljava.nio.channels.SelectornioSelector;
8. //broker和KafkaChannel(SocketChnnel)的映射
9. //这儿的kafkaChannel大家暂时可以理解为就是SocketChannel
10. //维护NodeId和KafkaChannel的映射关系
11. privatefinalMap<String,KafkaChannel>channels;
12. //记录已经完成发送的请求
13. privatefinalList<Send>completedSends;
14. //记录已经接收到的，并且处理完了的响应。
15. privatefinalList<NetworkReceive>completedReceives;
16. //已经接收到了，但是还没来得及处理的响应。
17. //一个连接，对应一个响应队列
18. privatefinalMap<KafkaChannel,Deque<NetworkReceive>>stagedReceives;
19. privatefinalSet<SelectionKey>immediatelyConnectedKeys;
20. //没有建立连接或者或者端口连接的主机
21. privatefinalList<String>disconnected;
22. //完成建立连接的主机
23. privatefinalList<String>connected;
24. //建立连接失败的主机。
25. privatefinalList<String>failedSends;
26. privatefinalTimetime;
27. privatefinalSelectorMetricssensors;
28. privatefinalStringmetricGrpPrefix;
29. privatefinalMap<String,String>metricTags;
30. //用于创建KafkaChannel的Builder
31. privatefinalChannelBuilderchannelBuilder;
32. privatefinalintmaxReceiveSize;
33. privatefinalbooleanmetricsPerConnection;
34. privatefinalIdleExpiryManageridleExpiryManager;

发起网络请求的第一步是连接、注册事件、发送、消息处理，涉及几个核心方法

1. 连接connect()方法

1. /**
2. *BeginconnectingtothegivenaddressandaddtheconnectiontothisnioSelectorassociatedwiththegivenid
3. *number.
4. *<p>
5. *Notethatthiscallonlyinitiatestheconnection,whichwillbecompletedonafuture{@link#poll(long)}
6. *call.Check{@link#connected()}toseewhich(ifany)connectionshavecompletedafteragivenpollcall.
7. *@paramidTheidforthenewconnection
8. *@paramaddressTheaddresstoconnectto
9. *@paramsendBufferSizeThesendbufferforthenewconnection
10. *@paramreceiveBufferSizeThereceivebufferforthenewconnection
11. *@throwsIllegalStateExceptionifthereisalreadyaconnectionforthatid
12. *@throwsIOExceptionifDNSresolutionfailsonthehostnameorifthebrokerisdown
13. */
14. @Override
15. publicvoidconnect(Stringid,InetSocketAddressaddress,intsendBufferSize,intreceiveBufferSize)throwsIOException{
16. if(this.channels.containsKey(id))
17. thrownewIllegalStateException("Thereisalreadyaconnectionforid"+id);
18. //获取到SocketChannel
19. SocketChannelsocketChannel=SocketChannel.open();
20. //设置为非阻塞的模式
21. socketChannel.configureBlocking(false);
22. Socketsocket=socketChannel.socket();
23. socket.setKeepAlive(true);
24. //设置网络参数，如发送和接收的buffer大小
25. if(sendBufferSize!=Selectable.USE_DEFAULT_BUFFER_SIZE)
26. socket.setSendBufferSize(sendBufferSize);
27. if(receiveBufferSize!=Selectable.USE_DEFAULT_BUFFER_SIZE)
28. socket.setReceiveBufferSize(receiveBufferSize);
29. //这个的默认值是false，代表要开启Nagle的算法
30. //它会把网络中的一些小的数据包收集起来，组合成一个大的数据包，再进行发送
31. //因为它认为如果网络中有大量的小的数据包在传输则会影响传输效率
32. socket.setTcpNoDelay(true);
33. booleanconnected;
34. try{
35. //尝试去服务器去连接，因为这儿非阻塞的
36. //有可能就立马连接成功，如果成功了就返回true
37. //也有可能需要很久才能连接成功，返回false。
38. connected=socketChannel.connect(address);
39. }catch(UnresolvedAddressExceptione){
40. socketChannel.close();
41. thrownewIOException("Can'tresolveaddress:"+address,e);
42. }catch(IOExceptione){
43. socketChannel.close();
44. throwe;
45. }
46. //SocketChannel往Selector上注册了一个OP_CONNECT
47. SelectionKeykey=socketChannel.register(nioSelector,SelectionKey.OP_CONNECT);
48. //根据SocketChannel封装出一个KafkaChannel
49. KafkaChannelchannel=channelBuilder.buildChannel(id,key,maxReceiveSize);
50. //把key和KafkaChannel关联起来
51. //我们可以根据key就找到KafkaChannel
52. //也可以根据KafkaChannel找到key
53. key.attach(channel);
54. //缓存起来
55. this.channels.put(id,channel);
57. //如果连接上了
58. if(connected){
59. //OP_CONNECTwon'ttriggerforimmediatelyconnectedchannels
60. log.debug("Immediatelyconnectedtonode{}",channel.id());
61. immediatelyConnectedKeys.add(key);
62. //取消前面注册OP_CONNECT事件。
63. key.interestOps(0);
64. }
65. }

2. 注册register()

1. /**
2. *RegisterthenioSelectorwithanexistingchannel
3. *Usethisonserver-side,whenaconnectionisacceptedbyadifferentthreadbutprocessedbytheSelector
4. *Notethatwearenotcheckingiftheconnectionidisvalid-sincetheconnectionalreadyexists
5. */
6. publicvoidregister(Stringid,SocketChannelsocketChannel)throwsClosedChannelException{
7. //往自己的Selector上面注册OP_READ事件
8. //这样的话，Processor线程就可以读取客户端发送过来的连接。
9. SelectionKeykey=socketChannel.register(nioSelector,SelectionKey.OP_READ);
10. //kafka里面对SocketChannel封装了一个KakaChannel
11. KafkaChannelchannel=channelBuilder.buildChannel(id,key,maxReceiveSize);
12. //key和channel
13. key.attach(channel);
14. //所以我们服务端这儿代码跟我们客户端的网络部分的代码是复用的
15. //channels里面维护了多个网络连接。
16. this.channels.put(id,channel);
17. }

3. 发送send()

1. /**
2. *Queuethegivenrequestforsendinginthesubsequent{@link#poll(long)}calls
3. *@paramsendTherequesttosend
4. */
5. publicvoidsend(Sendsend){
6. //获取到一个KafakChannel
7. KafkaChannelchannel=channelOrFail(send.destination());
8. try{
9. //重要方法
10. channel.setSend(send);
11. }catch(CancelledKeyExceptione){
12. this.failedSends.add(send.destination());
13. close(channel);
14. }
15. }

4. 消息处理poll()

1. @Override
2. publicvoidpoll(longtimeout)throwsIOException{
3. if(timeout<0)
4. thrownewIllegalArgumentException("timeoutshouldbe>=0");
5. //将上一次poll()方法返回的结果清空
6. clear();
7. if(hasStagedReceives()||!immediatelyConnectedKeys.isEmpty())
8. timeout=0;
9. /*checkreadykeys*/
10. longstartSelect=time.nanoseconds();
11. //从Selector上找到有多少个key注册了，等待I/O事件发生
12. intreadyKeys=select(timeout);
13. longendSelect=time.nanoseconds();
14. this.sensors.selectTime.record(endSelect-startSelect,time.milliseconds());
15. //上面刚刚确实是注册了一个key
16. if(readyKeys>0||!immediatelyConnectedKeys.isEmpty()){
17. //处理I/O事件，对这个Selector上面的key要进行处理
18. pollSelectionKeys(this.nioSelector.selectedKeys(),false,endSelect);
19. pollSelectionKeys(immediatelyConnectedKeys,true,endSelect);
20. }
21. //对stagedReceives里面的数据要进行处理
22. addToCompletedReceives();
23. longendIo=time.nanoseconds();
24. this.sensors.ioTime.record(endIo-endSelect,time.milliseconds());
25. //weusethetimeattheendofselecttoensurethatwedon'tcloseanyconnectionsthat
26. //havejustbeenprocessedinpollSelectionKeys
27. //完成处理后，关闭长链接
28. maybeCloseOldestConnection(endSelect);
29. }

5. 处理Selector上面的key

1. //用来处理OP_CONNECT,OP_READ,OP_WRITE事件，同时负责检测连接状态
2. privatevoidpollSelectionKeys(Iterable<SelectionKey>selectionKeys,
3. booleanisImmediatelyConnected,
4. longcurrentTimeNanos){
5. //获取到所有key
6. Iterator<SelectionKey>iterator=selectionKeys.iterator();
7. //遍历所有的key
8. while(iterator.hasNext()){
9. SelectionKeykey=iterator.next();
10. iterator.remove();
11. //根据key找到对应的KafkaChannel
12. KafkaChannelchannel=channel(key);
13. //registerallper-connectionmetricsatonce
14. sensors.maybeRegisterConnectionMetrics(channel.id());
15. if(idleExpiryManager!=null)
16. idleExpiryManager.update(channel.id(),currentTimeNanos);
17. try{
18. /*completeanyconnectionsthathavefinishedtheirhandshake(eithernormallyorimmediately)*/
19. //处理完成连接和OP_CONNECT的事件
20. if(isImmediatelyConnected||key.isConnectable()){
21. //完成网络的连接。
22. if(channel.finishConnect()){
23. //网络连接已经完成了以后，就把这个channel添加到已连接的集合中
24. this.connected.add(channel.id());
25. this.sensors.connectionCreated.record();
26. SocketChannelsocketChannel=(SocketChannel)key.channel();
27. log.debug("CreatedsocketwithSO_RCVBUF={},SO_SNDBUF={},SO_TIMEOUT={}tonode{}",
28. socketChannel.socket().getReceiveBufferSize(),
29. socketChannel.socket().getSendBufferSize(),
30. socketChannel.socket().getSoTimeout(),
31. channel.id());
32. }else
33. continue;
34. }
35. /*ifchannelisnotreadyfinishprepare*/
36. //身份认证
37. if(channel.isConnected()&&!channel.ready())
38. channel.prepare();
39. /*ifchannelisreadyreadfromanyconnectionsthathavereadabledata*/
40. if(channel.ready()&&key.isReadable()&&!hasStagedReceive(channel)){
41. NetworkReceivenetworkReceive;
42. //处理OP_READ事件，接受服务端发送回来的响应（请求）
43. //networkReceive代表的就是一个服务端发送回来的响应
44. while((networkReceive=channel.read())!=null)
45. addToStagedReceives(channel,networkReceive);
46. }
47. /*ifchannelisreadywritetoanysocketsthathavespaceintheirbufferandforwhichwehavedata*/
48. //处理OP_WRITE事件
49. if(channel.ready()&&key.isWritable()){
50. //获取到要发送的那个网络请求，往服务端发送数据
51. //如果消息被发送出去了，就会移除OP_WRITE
52. Sendsend=channel.write();
53. //已经完成响应消息的发送
54. if(send!=null){
55. this.completedSends.add(send);
56. this.sensors.recordBytesSent(channel.id(),send.size());
57. }
58. }
59. /*cancelanydefunctsockets*/
60. if(!key.isValid()){
61. close(channel);
62. this.disconnected.add(channel.id());
63. }
64. }catch(Exceptione){
65. Stringdesc=channel.socketDescription();
66. if(einstanceofIOException)
67. log.debug("Connectionwith{}disconnected",desc,e);
68. else
69. log.warn("Unexpectederrorfrom{};closingconnection",desc,e);
70. close(channel);
71. //添加到连接失败的集合中
72. this.disconnected.add(channel.id());
73. }
74. }
75. }

2.2 MetadataUpdater

是NetworkClient用于请求更新集群元数据信息并检索集群节点的接口，是一个非线程安全的内部类，有两个实现类，分别是DefaultMetadataUpdater和ManualMetadataUpdater，NetworkClient用到的是DefaultMetadataUpdater类，是NetworkClient的默认实现类，同时是NetworkClient的内部类，从源码可以看到，如下。

1. if(metadataUpdater==null){
2. if(metadata==null)
3. thrownewIllegalArgumentException("`metadata`mustnotbenull");
4. this.metadataUpdater=newDefaultMetadataUpdater(metadata);
5. }else{
6. this.metadataUpdater=metadataUpdater;
7. }
8. ……
9. classDefaultMetadataUpdaterimplementsMetadataUpdater
10. {
11. /*thecurrentclustermetadata*/
12. //集群元数据对象
13. privatefinalMetadatametadata;
14. /*trueifthereisametadatarequestthathasbeensentandforwhichwehavenotyetreceivedaresponse*/
15. //用来标识是否已经发送过MetadataRequest，若是已发送，则无需重复发送
16. privatebooleanmetadataFetchInProgress;
17. /*thelasttimestampwhennobrokernodeisavailabletoconnect*/
18. //记录没有发现可用节点的时间戳
19. privatelonglastNoNodeAvailableMs;
20. DefaultMetadataUpdater(Metadatametadata){
21. this.metadata=metadata;
22. this.metadataFetchInProgress=false;
23. this.lastNoNodeAvailableMs=0;
24. }
25. //返回集群节点集合
26. @Override
27. publicList<Node>fetchNodes(){
28. returnmetadata.fetch().nodes();
29. }
30. @Override
31. publicbooleanisUpdateDue(longnow){
32. return!this.metadataFetchInProgress&&this.metadata.timeToNextUpdate(now)==0;
33. }
34. //核心方法，判断当前集群保存的元数据是否需要更新，如果需要更新则发送MetadataRequest请求
35. @Override
36. publiclongmaybeUpdate(longnow){
37. //shouldweupdateourmetadata?
38. //获取下次更新元数据的时间戳
39. longtimeToNextMetadataUpdate=metadata.timeToNextUpdate(now);
40. //获取下次重试连接服务端的时间戳
41. longtimeToNextReconnectAttempt=Math.max(this.lastNoNodeAvailableMs+metadata.refreshBackoff()-now,0);
42. //检测是否已经发送过MetadataRequest请求
43. longwaitForMetadataFetch=this.metadataFetchInProgress?Integer.MAX_VALUE:0;
44. //ifthereisnonodeavailabletoconnect,backoffrefreshingmetadata
45. longmetadataTimeout=Math.max(Math.max(timeToNextMetadataUpdate,timeToNextReconnectAttempt),
46. waitForMetadataFetch);
47. if(metadataTimeout==0){
48. //Bewarethatthebehaviorofthismethodandthecomputationoftimeoutsforpoll()are
49. //highlydependentonthebehaviorofleastLoadedNode.
50. //找到负载最小的节点
51. Nodenode=leastLoadedNode(now);
52. //创建MetadataRequest请求，待触发poll（）方法执行真正的发送操作。
53. maybeUpdate(now,node);
54. }
55. returnmetadataTimeout;
56. }
57. //处理没有建立好连接的请求
58. @Override
59. publicbooleanmaybeHandleDisconnection(ClientRequestrequest){
60. ApiKeysrequestKey=ApiKeys.forId(request.request().header().apiKey());
61. if(requestKey==ApiKeys.METADATA&&request.isInitiatedByNetworkClient()){
62. Clustercluster=metadata.fetch();
63. if(cluster.isBootstrapConfigured()){
64. intnodeId=Integer.parseInt(request.request().destination());
65. Nodenode=cluster.nodeById(nodeId);
66. if(node!=null)
67. log.warn("Bootstrapbroker{}:{}disconnected",node.host(),node.port());
68. }
69. metadataFetchInProgress=false;
70. returntrue;
71. }
72. returnfalse;
73. }
74. //解析响应信息
75. @Override
76. publicbooleanmaybeHandleCompletedReceive(ClientRequestreq,longnow,Structbody){
77. shortapiKey=req.request().header().apiKey();
78. //检查是否为MetadataRequest请求
79. if(apiKey==ApiKeys.METADATA.id&&req.isInitiatedByNetworkClient()){
80. //处理响应
81. handleResponse(req.request().header(),body,now);
82. returntrue;
83. }
84. returnfalse;
85. }
86. @Override
87. publicvoidrequestUpdate(){
88. this.metadata.requestUpdate();
89. }
90. //处理MetadataRequest请求响应
91. privatevoidhandleResponse(RequestHeaderheader,Structbody,longnow){
92. this.metadataFetchInProgress=false;
93. //因为服务端发送回来的是一个二进制的数据结构
94. //所以生产者这儿要对这个数据结构要进行解析
95. //解析完了以后就封装成一个MetadataResponse对象。
96. MetadataResponseresponse=newMetadataResponse(body);
97. //响应里面会带回来元数据的信息
98. //获取到了从服务端拉取的集群的元数据信息。
99. Clustercluster=response.cluster();
100. //checkifanytopicsmetadatafailedtogetupdated
101. Map<String,Errors>errors=response.errors();
102. if(!errors.isEmpty())
103. log.warn("Errorwhilefetchingmetadatawithcorrelationid{}:{}",header.correlationId(),errors);
104. //don'tupdatetheclusteriftherearenovalidnodes…thetopicwewantmaystillbeintheprocessofbeing
105. //createdwhichmeanswewillgeterrorsandnonodesuntilitexists
106. //如果正常获取到了元数据的信息
107. if(cluster.nodes().size()>0){
108. //更新元数据信息。
109. this.metadata.update(cluster,now);
110. }else{
111. log.trace("Ignoringemptymetadataresponsewithcorrelationid{}.",header.correlationId());
112. this.metadata.failedUpdate(now);
113. }
114. }
115. /**
116. *Createametadatarequestforthegiventopics
117. */
118. privateClientRequestrequest(longnow,Stringnode,MetadataRequestmetadata){
119. RequestSendsend=newRequestSend(node,nextRequestHeader(ApiKeys.METADATA),metadata.toStruct());
120. returnnewClientRequest(now,true,send,null,true);
121. }
122. /**
123. *Addametadatarequesttothelistofsendsifwecanmakeone
124. */
125. privatevoidmaybeUpdate(longnow,Nodenode){
126. //检测node是否可用
127. if(node==null){
128. log.debug("Giveupsendingmetadatarequestsincenonodeisavailable");
129. //markthetimestampfornonodeavailabletoconnect
130. this.lastNoNodeAvailableMs=now;
131. return;
132. }
133. StringnodeConnectionId=node.idString();
134. //判断网络连接是否应建立好，是否可用向该节点发送请求
135. if(canSendRequest(nodeConnectionId)){
136. this.metadataFetchInProgress=true;
137. MetadataRequestmetadataRequest;
138. //指定需要更新元数据的topic
139. if(metadata.needMetadataForAllTopics())
140. //封装请求（获取所有topics）的元数据信息的请求
141. metadataRequest=MetadataRequest.allTopics();
142. else
143. //我们默认走的这儿的这个方法
144. //就是拉取我们发送消息的对应的topic的方法
145. metadataRequest=newMetadataRequest(newArrayList<>(metadata.topics()));
146. //这儿就给我们创建了一个请求（拉取元数据的）
147. ClientRequestclientRequest=request(now,nodeConnectionId,metadataRequest);
148. log.debug("Sendingmetadatarequest{}tonode{}",metadataRequest,node.id());
149. //缓存请求，待下次触发poll（）方法执行发送操作
150. doSend(clientRequest,now);
151. }elseif(connectionStates.canConnect(nodeConnectionId,now)){
152. //wedon'thaveaconnectiontothisnoderightnow,makeone
153. log.debug("Initializeconnectiontonode{}forsendingmetadatarequest",node.id());
154. //初始化连接
155. initiateConnect(node,now);
156. //IfinitiateConnectfailedimmediately,thisnodewillbeputintoblackoutandwe
157. //shouldallowimmediatelyretryingincasethereisanothercandidatenode.Ifit
158. //isstillconnecting,theworstcaseisthatweendupsettingalongertimeout
159. //onthenextroundandthenwaitfortheresponse.
160. }else{//connected,butcan'tsendmoreORconnecting
161. //Ineithercase,wejustneedtowaitforanetworkeventtoletusknowtheselected
162. //connectionmightbeusableagain.
163. this.lastNoNodeAvailableMs=now;
164. }
165. }
166. }
167. 将ClientRequest请求缓存到InFlightRequest缓存队列中。
168. privatevoiddoSend(ClientRequestrequest,longnow){
169. request.setSendTimeMs(now);
170. //这儿往inFlightRequests缓存队列里面还没有收到响应的请求,默认最多能存5个请求
171. this.inFlightRequests.add(request);
172. //然后不断调用Selector的send（）方法
173. selector.send(request.request());
174. }

2.3 InFlightRequests

这个类是一个请求队列，用于缓存已经发送出去但是没有收到响应的ClientRequest，提供了许多管理缓存队列的方法，支持通过配置参数控制ClientRequest的数量，通过源码可以看到其底层数据结构Map。

1. /**
2. *Thesetofrequestswhichhavebeensentorarebeingsentbuthaven'tyetreceivedaresponse
3. */
4. finalclassInFlightRequests{
5. privatefinalintmaxInFlightRequestsPerConnection;
6. privatefinalMap<String,Deque<ClientRequest>>requests=newHashMap<>();
7. publicInFlightRequests(intmaxInFlightRequestsPerConnection){
8. this.maxInFlightRequestsPerConnection=maxInFlightRequestsPerConnection;
9. }
10. ……
11. }

除了包含了很多关于处理队列的方法之外，有一个比较重要的方法着重看一下canSendMore()。

1. /**
2. *Canwesendmorerequeststothisnode?
3. *
4. *@paramnodeNodeinquestion
5. *@returntrueiffwehavenorequestsstillbeingsenttothegivennode
6. */
7. publicbooleancanSendMore(Stringnode){
8. //获得要发送到节点的ClientRequest队列
9. Deque<ClientRequest>queue=requests.get(node);
10. //如果节点出现请求堆积，未及时处理，则有可能出现请求超时的情况
11. returnqueue==null||queue.isEmpty()||
12. (queue.peekFirst().request().completed()
13. &&queue.size()<this.maxInFlightRequestsPerConnection);
14. }

了解完上面几个核心类之后，我们开始剖析NetworkClient的流程和实现。

2.4 NetworkClient

Kafka中所有的消息都都需要借助NetworkClient与上下游建立发送通道，其重要性不言而喻。这里我们只考虑消息成功的流程，异常处理不做解析，相对而言没那么重要，消息发送的流程大致如下：

1. 首先调用ready()方法，判断节点是否具备发送消息的条件

2. 通过isReady()方法判断是否可以往节点发送更多请求，用来检查是否有请求堆积

3. 使用initiateConnect初始化连接

4. 然后调用selector的connect()方法建立连接

5. 获取SocketChannel，与服务端建立连接

6. SocketChannel往Selector注册OP_CONNECT事件

7. 调用send()方式发送请求

8. 调用poll()方法处理请求

下面就根据消息发送流程涉及的核心方法进行剖析，了解每个流程中涉及的主要操作。

1. 检查节点是否满足消息发送条件

1. /**
2. *Beginconnectingtothegivennode,returntrueifwearealreadyconnectedandreadytosendtothatnode.
3. *
4. *@paramnodeThenodetocheck
5. *@paramnowThecurrenttimestamp
6. *@returnTrueifwearereadytosendtothegivennode
7. */
8. @Override
9. publicbooleanready(Nodenode,longnow){
10. if(node.isEmpty())
11. thrownewIllegalArgumentException("Cannotconnecttoemptynode"+node);
12. //判断要发送消息的主机，是否具备发送消息的条件
13. if(isReady(node,now))
14. returntrue;
15. //判断是否可以尝试去建立网络
16. if(connectionStates.canConnect(node.idString(),now))
17. //ifweareinterestedinsendingtoanodeandwedon'thaveaconnectiontoit,initiateone
18. //初始化连接
19. //绑定了连接到事件而已
20. initiateConnect(node,now);
21. returnfalse;
22. }

2. 初始化连接

1. /**
2. *Initiateaconnectiontothegivennode
3. */
4. privatevoidinitiateConnect(Nodenode,longnow){
5. StringnodeConnectionId=node.idString();
6. try{
7. log.debug("Initiatingconnectiontonode{}at{}:{}.",node.id(),node.host(),node.port());
8. this.connectionStates.connecting(nodeConnectionId,now);
9. //开始建立连接
10. selector.connect(nodeConnectionId,
11. newInetSocketAddress(node.host(),node.port()),
12. this.socketSendBuffer,
13. this.socketReceiveBuffer);
14. }catch(IOExceptione){
15. /*attemptfailed,we'lltryagainafterthebackoff*/
16. connectionStates.disconnected(nodeConnectionId,now);
17. /*maybetheproblemisourmetadata,updateit*/
18. metadataUpdater.requestUpdate();
19. log.debug("Errorconnectingtonode{}at{}:{}:",node.id(),node.host(),node.port(),e);
20. }
21. }

3. initiateConnect()方法中调用的connect()方法就是Selectable实现类Selector的connect()方法，包括获取SocketChannel并注册OP_CONNECT、OP_READ、 OP_WRITE事件上面已经分析了，这里不做赘述，完成以上一系列建立网络连接的动作之后将消息请求发送到下游节点，Sender的send()方法会调用NetworkClient的send()方法进行发送，而NetworkClient的send()方法最终调用了Selector的send()方法。

1. /**
2. *Queueupthegivenrequestforsending.Requestscanonlybesentouttoreadynodes.
3. *
4. *@paramrequestTherequest
5. *@paramnowThecurrenttimestamp
6. */
7. @Override
8. publicvoidsend(ClientRequestrequest,longnow){
9. StringnodeId=request.request().destination();
10. //判断已经建立连接状态的节点能否接收更多请求
11. if(!canSendRequest(nodeId))
12. thrownewIllegalStateException("Attempttosendarequesttonode"+nodeId+"whichisnotready.");
13. //发送ClientRequest
14. doSend(request,now);
15. }
16. privatevoiddoSend(ClientRequestrequest,longnow){
17. request.setSendTimeMs(now);
18. //缓存请求
19. this.inFlightRequests.add(request);
20. selector.send(request.request());
21. }

4. 最后调用poll()方法处理请求

1. /**
2. *Doactualreadsandwritestosockets.
3. *
4. *@paramtimeoutThemaximumamountoftimetowait(inms)forresponsesiftherearenoneimmediately,
5. *mustbenon-negative.Theactualtimeoutwillbetheminimumoftimeout,requesttimeoutand
6. *metadatatimeout
7. *@paramnowThecurrenttimeinmilliseconds
8. *@returnThelistofresponsesreceived
9. */
10. @Override
11. publicList<ClientResponse>poll(longtimeout,longnow){
13. //步骤一：请求更新元数据
14. longmetadataTimeout=metadataUpdater.maybeUpdate(now);
15. try{
17. //步骤二：执行I/O操作，发送请求
18. this.selector.poll(Utils.min(timeout,metadataTimeout,requestTimeoutMs));
19. }catch(IOExceptione){
20. log.error("UnexpectederrorduringI/O",e);
21. }
22. //processcompletedactions
23. longupdatedNow=this.time.milliseconds();
24. List<ClientResponse>responses=newArrayList<>();
25. //步骤三：处理各种响应
26. handleCompletedSends(responses,updatedNow);
27. handleCompletedReceives(responses,updatedNow);
28. handleDisconnections(responses,updatedNow);
29. handleConnections();
30. handleTimedOutRequests(responses,updatedNow);
31. //invokecallbacks
32. //循环调用ClientRequest的callback回调函数
33. for(ClientResponseresponse:responses){
34. if(response.request().hasCallback()){
35. try{
36. response.request().callback().onComplete(response);
37. }catch(Exceptione){
38. log.error("Uncaughterrorinrequestcompletion:",e);
39. }
40. }
41. }
42. returnresponses;
43. }

有兴趣的童鞋可以继续深究回调函数的逻辑和Selector的操作。

补充说明：

上面有一个点没有涉及到，就是kafka的内存池，可以去看一下BufferPool这个类，这一块知识点应该是要在上一篇文章说的，突然想起来漏掉了，在这里做一下补充，对应就是我们前面说到的RecordAccumulator这个类的数据结构，封装好RecordAccumulator对象是有多个Dqueue组成，每个Dqueue由多个RecordBatch组成，除此之外，RecordAccumulator还包括了BufferPool内存池，这里再稍微回忆一下，RecordAccumulator类初始化了ConcurrentMap。

1. publicfinalclassRecordAccumulator{
2. ……
3. privatefinalBufferPoolfree;
4. ……
5. privatefinalConcurrentMap<TopicPartition,Deque<RecordBatch>>batches;
6. }

如图所示，我们重点关于内存的分配allocate()和释放deallocate()这个两个方法，有兴趣的小伙伴可以私底下看一下，这个类的代码总共也就三百多行，内容不是很多，欢迎一起交流学习，这里就不做展开了，免得影响本文的主题。

3. 小结

本文主要是剖析Kafka生产者发送消息的真正执行者Sender线程以及作为消息上下游的传输通道NetworkClient组件，主要涉及到NIO的应用，同时介绍了发送消息主要涉及的核心依赖类。写这篇文章主要是起到一个承上启下的作用，既是对前面分析Kafka生产者发送消息的补充，同时也为接下来剖析消费者消费上游消息作铺垫，写得有点不成体系，写文章的思路主要考虑用总分的思路作为线索去分析，个人觉得篇幅过长不方便阅读，所以会尽量精简，重点分析核心方法和流程，希望对读者有所帮助。

原文链接：https://mp.weixin.qq.com/s/_xjxjahAp76Oft05LYYYCA