4.2.3 网络客户端轮询

悠悠 2023-01-13 15:55 31阅读 0赞

--------------------

#### 4.2.3 网络客户端轮询 ####

> 本节一开始提到消费者通过消费者网络客户端对象(ConsumerNetworkClient)这中间花了很大篇幅去分析异步请求。现在再回到消费者网络客户端发送请求之后的流程，即消费者网络客户端的轮询。

1.  轮询发送请求客户端发送请求后，但又不知道什么时候服务端才返回结果。客户端获取结果的3种轮询方法如下。

*  客户端不想等待，设置超时时间为0，表示请求发送完就立即回到调用者的主线程中。
 *  指定了超时时间，如果在指定时间内没有结果，最后也要返回到调用者的主线程中。
 *  设置超时时间为最大值，表示除非有结果返回，否则一直阻塞，永远不会回到主线程。

相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

表4-7列举了消费者网络客户端3种不同参数的轮询方法，分别对应上面的3种情景。

![在这里插入图片描述][20210408222710765.png]

先来复习一下普通的网络客户端(NetworkCl1ent)发送一个请求的过程，它要经过下面的3个步骤。

(1)clientready(node)。客户端连接上目标节点，并已经准备好发送请求。
    (2)cHentsend(request)。发送请求，只将请求设置到网络通道，下一步才真正发送。
    (3)cHentpoll(t1111eout)。客户端轮询获取结果，如果有回调方法会处理响应结果。

消费者的网络客户端底层对NetworkCl1ent进行了封装，它完成一次请求发送的步骤如下。

(1)send()方法创建客户端请求并缓存到未发送的请求集(每个节点对应一个请求列表)。
    (2)poll()方法会对未发迭的请求集，迭代每个节点的每个客户端请求。
    (3)trySend()方法会调用NetworkCl1entsend()，暂存请求到网络通道。
    (4)cHentPoll()方法会调用NetworkCHnetpoll()，真正把请求发迭出去。

消费者的网络客户端尝试发送请求时，会循环处理未发送的请求集(unsent字典)。如果客户端已经和对应的服务端节点建立连接，就可以将客户端请求通过clientsend()设置到对应的网络通道中，调用clientpoll()把网络通道巾的客户端请求真正发送出去。

因为调用一次Networkc:t1ent的轮询方法可以发送多个请求，所以不需要在每次调用clientsend()后者11立即调用clientpoll()。而是应把所有准备好的客户端请求都设置到对应的网络通道，最后才执行一次轮询。另外，在调用clientsend()将客户端请求设置到网络通道后，可以将其从未发送的请求集中删除。下次再调用尝试发送请求时，客户端请求不会驻留在未发送的请求集字典中，不会l材现重复发迭的情况。相关代码如下·  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

1.  定时任务

在Java中实现定时任务有几种方式：使用循环和睡眠方式实现的普通线程、单线程的Timer和TimerTask、线程池的ScheduledExecutorService。用Java实现定时任务，一般这种任务会无条件定时执行，Kafka中的定时任务有下面几个特点。

*  定时任务是通过消费者的轮询动作触发的，如果没有轮询，就不会执行定时任务。
 *  定时任务也会发送请求给服务端，并且在收到响应结果后需要做一些具体的业务处理。
 *  如果上一次的任务没有完成，定时任务就不会启动新的任务。

Kafka用轮询和队列实现了定时任务，并且引入了延迟任务的概念。延迟任务指的是：当前任务完成后，记录下一次任务被调度的时间，并且以下次调度的时间创建一个新的延迟任务。随着时间的流逝，在某个时刻，如果任务的调度时间超过当前时刻，就说明之前创建的延迟任务的调度时间已经到了，不能再延迟了，需要立即执行。客户端决定任务什么时候开始调度有下面两种方式。

*  记录下一次调度的绝对时间，延迟的任务会在记录的这个时间点开始执行。
 *  指定间隔多长时间后再；欠执行，延迟的任务在当前任务执行后，再经过这个时间间隔开始执行。

方式一比较容易实现，只要用任务的调度时间和当前时间比较即可。如果调度时间超过当前时间，就说明需要执行任务。方式二要保存上一个任务的调度时间，然后用当前时间减去上一次的调度时间得到差距时间，将其再和间隔时间比较，才能判断是否可以执行。定时任务与延迟任务两个概念并不矛盾，定时任务指的是任务会被定时执行，延迟任务指的是任务会被延迟执行，并且会以定时的方式被调度执行。延迟任务需要通过一个外部调用者不停地检测任务的调度是否已经到了。如果只是把延迟任务放入任务队列，却没有外部调用者的定时检测，那么延迟的任务还是没有机会执行。

在客户端轮询的任何时刻，系统中只会存在一个延迟的任务，这个任务只会有3种状态：等待执行、正在执行、执行完成。如果当前任务执行完，它的生命周期就彻底结束了，并且会创建出一个新的任务，这个新任务的状态也会经历等待执行、正在执行、执行完成这3种状态。例如，客户端每隔10秒向协调者发送1次心跳，第一次任务执行完的时间点是2016-07-0622:54:35。任务(！)执行完成后，它的生命周期就结束了，并且会新创建一个调度时间点为2日16-07-0622:54：“的延迟任务(2)，然后将这个新创建的延迟任务(2)加入延迟任务队列(delayedTasks)。同时，外部的Kafka客户端每隔l秒执行l次轮询，每次轮询会检查延迟任务队列中是否有超时的任务需要取出来执行。由于轮询的时间间隔为l秒，经过10次轮询后，当前时间会大于延迟任务(2)的调度时间，客户端就会开始执行延迟任务(2)。

注意：因为定时任务的执行是由客户端轮询触发的，所以最好保证客户端轮询的间隔时间比定时任务的间隔时间长。这样定时的任务就不会被延迟太久才轮询到。假设客户端轮询问隔为5秒，那么定时任务的间隔要设直成大于5秒。否则，如果定时任务的问隔为l秒，客户端5秒才轮询一次，那么定时任务就被延迟了4秒才开始执行。

1.  延迟的任务队列

客户端要调度定时任务，需要调用消费者网络客户端的schedule()方法，并且指定延迟的任务和调用的时间点。客户端有多种定时任务，会使用队列来保存所有需要定时执行的延迟任务。这个队列是一个按照任务的调度时间进行排序的优先级队列。

客户端调用schedule()方法会将任务添加到优先级队列。优先级队列总是有序的，并且眼任务加入队列的时间没有关系。例如，先加入队列的任务(1)超时时间为10秒，后加入队列的任务(2)超时时间为5秒，队列第一个元素是任务(2)，因为任务。)的超时时间比任务(1)要小。优先级队列中的延迟任务按照时间顺序排序，即最先应该执行的任务放在队列头部。那么从队列拉出来的第一个任务一定是最早到期的，后面任务的到期时间一定比第一个任务要晚。每个任务的调度时间在放入队列时就被固定好了，而且一般是在当前时间之后。

延迟任务的弹1+1方法(poll())指定了参数当前时间(now)，表示会依次弹出所有调度时间小于当前时间的任务。由于客户端会不断轮询，参数now也会不断增加。在某个时刻一定会有调度时间满足now大于等于队列第一个任务的，表示这个任务到期了应该开始执行。也有可能在这个时刻有多个任务同时满足这个条件，那么它们都会被选中执行。例如，队列中每个任务的调度时间分别是Task2=2秒、Task1=3秒、Task5=5秒、Task3=7衫’~，Task4=9秒，这里任务的编号不一定有序，但是调度时间是有序的。如果当前时间是l秒，不会弹出任何任务；当前时间是2秒，则只会弹出Task2；当前时间是8秒，则会接着弹：·HTasks,Task3。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

放在队列中的延迟任务只能通过外部调用轮询方法的方式才有机会执行，而队列的轮询方法只通过ConsumerNetworkClientpoll()调用。下面是延迟队列poll()方法的调用战：

![在这里插入图片描述][20210408222845794.png]

延迟类型的任务(DelayedTask)在消费者客户端有两个实现类：心跳任务(AbstractCoordinator.HeartbeatTask)和定时提交任务(AbstractCoordinator.AutoCommitTask)。这两种任务者IS属于需要定时执行的任务，由于都和消费组相关，所以会通过“消费者的协调者”对象，发送对应的客户端请求给服务端的协调者节点。

*  HeartbeatTask。消费者周期性地向协调者发送心跳，表示向己是存活的。
 *  AutoCOMMitTask。消费者定期地向协调者发送“偏移量提交请求”，保存消费进度。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/c6faa533b2a0424b9b3e33917ca444ea.png
[20210408222710765.png]: /images/20221022/9a647d878d574fbd9c8c02231100def8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/eab4671af4194ccbb73be369f27ffa32.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/e14c35128b09400d90eb753430d4b0bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/9a7c1086f27b4dafbea2dda3f519d52d.png
[20210408222845794.png]: /images/20221022/1e54251f2a844408a82db1a3010069ce.png