4.2.2异步请求高级模式

ゝ一世哀愁。 2023-01-13 15:55 75阅读 0赞

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

#### 4.2.2异步请求高级模式 ####

有时候客户端响应对象中的数据比较简单，可能就只有一条数据，而且类型也是确定的。比如获取分区偏移量，客户端响应结果只有一条数据，类型为Long。我们可能希望从异步请求得到的结果就是这个偏移盘，而不希望根据Cl1.entResponse对象再去获取。异步请求对象提供了“组合加适配器”(co111pose+Adapter)的模式，可以让调用者直接获取异步请求对象的结果。

1.  异步请求适配器

异步请求为客户端提供调用自定义业务处理逻辑的人口，除了“添加监昕器”的方式，还有一种是“组合加适配器”模式。这种模式不仅有监听器，还能够对客户端响应结果做一次转换。这个转换操作实际上是对客户端响应结果进行解析，直接返回客户端想要的数据。注意：“组合加适配器”模式中，“组合”表示组装一个监听器，“运配器”表示对客户端响应结果进行适配。

下面比较一下组合模式和上一节两种“获取客户端响应结果”模式(普通模式)的不同点。

(1)普通模式使用监昕器，并将异步请求的结果传给监听器的回调方法。组合模式也有监听器，但对异步请求的结果做了一次转换。
    (2)普通模式不使用监昕器，直接获取异步请求结果。组合模式也获取异步请求的结果，但有监听器。
    (3)普通模式在获取到异步请求的结果后，才会执行回调处理。但组合模式在获取到异步请求的结果后，流程就结束了。组合模式的回调处理是在适配器的回调方法中完成的，适配器转换后的结果就是异步请求的最终结果。

下面以客户端发送LIST\_OFFSETS请求获取分区的偏移量为例，说明“组合加适配器”模式的使用方法。为了和普通模式的用法进行对比，这里也贴出了采用监听器模式发送拉取请求的用法：

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

如表4-6所示，拉取请求使用添加监昕器的方式，回调方法的参数只有客户端响应。获取偏移量请求使用组合的方式，回调方法除了客户端响应，还有一个异步请求对象。

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

比较一下异步请求监听器和异步请求适配器的回调方法差异，后者多了一个异步请求对象。以LIST\_OFFSETS请求为例，适配器有两个泛型参数，分别是T：口i.entResponse、S:Long。相关代码如下：  
![在这里插入图片描述][20210408220603803.png]

如图4-28所示，对于适配器的两个类型，T和s可以理解为输入类型T和输出类型5。适配器的作用是获取输入类型T=Cli.entResponse，将数据填充到输出类型S=Long中，这也是适配器抽象类的回调方法需要传入两个参数的原因。

注意：适配器可以理解为Scala的map()函数，比如U.st(l,2,3).map(i.=>i…toStri.ng)。输入类型是T=Integer，输出类型是S=Stri.ng,!'lap()函数的返回值类型是Li.st\[S］。

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

要将输入类型转为输州类型，使用普通异步请求的方式来做的话，需要先获取到异步请求的客户端响应结果，然后将其转为输出类型。如果使用组合模式，异步请求结果的返回值直接就是输出类型。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]

下面先分析组合模式的一种错误的设计方案，再看一下Kafka是如何使用组合模式对结果进行转换的。

1.  异步请求的组合模式  
    “组合加适配器”模式的目的是将输入类型转换为输出类型。假设客户端发送请求返回的是输入类型的异步请求，可以为异步请求添加一个监听器，在监昕器的回调方法中，将输入类型的结果转换为输~H类型。相关代码如下：

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

但这里存在的问题是：回调方法转换后的结果，它的作用域对于客户端是不可见的，所以外部客户端无法获取转换后的结果。解决办法是再引人一个新的异步请求对象，这个新异步请求保存的是转换后的结果，然后将其暴露给客户端。客户端获取新异步请求对象的结果，就是转换后的输出类型数据。相关代码如下：

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

异步请求对象compose()方法的设计思路，采用了第二种方案实现将输入类型转为输出类型。组合模式的调用链比较复杂，还是以LIST\_OFFSETS请求为例。适配器类的两个泛型＜T,S＞分别表示输入和输出，对应这里的示例，输入类型T＝口i.entResponse，输出类型S=Long。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

如图4-29所示，拉取器调用发送请求，再加上组合方法返回的是一个新创建的异步请求［町，然后客户端会轮询新异步请求。新异步请求完成后，拉取器获取这个新异步请求对象的结果就是转换后的结果，它的类型是S=Long，表示分区的偏移量。整个调用流程的具体步骤如下。

(1)客户端发送请求，返回第一个异步请求［T］。
    (2)在第一个请求［T］上组合适配器后，返回第二个异步请求［S］。
    (3)为第一异步请求［T］添加一个监昕器。
    (4)客户端轮询到响电结果，异步请求［T］完成，会调用监昕器的o白cce叫)回调方法。
    (5)在监听器的回调实现中，会调用适配器的onSuccess()回调方法。
    (6)在适配器的回调实现中，会将异步请求［T］的结果转为［呵，并完成异步请求［S］。
    (7)客户端获取异步请求［S］的的结果。

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

如图4-30所示，“组合加适配器”模式比普通的异步请求模式多了右侧的3个对象，具体步骤如下。

(1)客户端发送请求得到的是旧异步请求，类型为［T］，实际上是一个客户端响应。
    (2)在旧的异步请求上调用组合方法，返回一个新的异步请求，类型为［SJ，可以是严｜定义的类型。
    (3)在旧异步请求对象上添加一个监昕器，后续当它完成后，会调用适配器的回调方法。
    (4)异步请求也是一个异步请求完成的处理器，当客户端轮询到结果，调用其onComplete()方法。
    (5)异步请求完成处理器的回调方法会调用异步请求的complete()方法。
    (6)异步请求完成后，会调用步骤(3)中监昕器的onSuccess()回调方法。
    (7)监昕器的回调方法会调用组合方法传入适配器的onSuccess()回调方法。
    (8)在适配器的回调方法中，会解析客户端响应结果，完成步骤(2)的异步请求。
    (9)客户端获取步骤(2)异步请求的结果值，流程结束。

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

总结一下。客户端发送请求得到的是旧异步请求，在旧异步请求上调用组合方法，返回的是新异步请求。客户端轮询时会先触发旧异步请求中监听器的回调方法然后才会调用适配器的回调方法。适配器的回调处理是将旧异步请求的客户端响应结果转换成新的输归类型，最后设置到新异步请求的结果中。

1.  组合模式的运用：获取偏移量

如果抛开异步请求、回调方法、监听器这些概念，只看LIST\_OFFSETS请求的业务处理逻辑，获取分区的偏移量需要指定一个时间戳，(分区的主副本)服务端节点的日志文件会返回这个时间戳之前所有片段文件的基准偏移量。如果是消费者要获取分区最近提交的偏移盆，但消费者对应的协调节点没有保存这个信息，消费者只能根据客户端设置的重置策略，去分区的主副本节点读取一个偏移孟值。获取分区的偏移量方法对异步请求的使用，和消费者拉取请求有如下两点区别。

*  客户端轮询的参数不是超时时间，而是异步请求：client.poll(futur的。
 *  最外层使用了死循环，确保只有在异步请求成功完成时，整个方法才会结束。

调用clientpoll(future)与调用clientpoll(tif11eout)是不同的：后者不管异步请求有没有完成，都会在给定的超时时间内返回，因此在轮询后调用异步请求的结果不一定有值；而前者必须等到异步请求完成了才会结束，在轮询结束后可以获取异步请求的值。带有异步请求参数的轮询方法本身也有一层循环，li.stOffset()方法实际上有两层循环。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]

正常情况下，客户端发送“获取分区的偏移量”请求，能够获取到响应结果，那么异步请求执行成功且有值。但如果服务端没有这个分区，或者分区还没有主副本，就不需要发送异步请求。这里会构造一个失败的异步请求，返回至Uli.stOffset()主方法中。主方法判断异步请求虽然完成，但却是失败的，就要隔段时间再重试。为了保证请求能够被执行，如果在发送请求之前出现异常，1需要处理错误并重试。相关代码如下：

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

异步请求完成有两种情况：成功完成和失败完成。上面两种异常情况构造的异步请求虽然完成了，但却是失败的。为什么不把异常情况的异步请求认为没有完成呢？下面代码段第一部分异常情况的异步请求对象，如果是未完成的，就会一直阻塞在内层的循环中，这样根本就没有机会执行发送请求的逻辑。

如果是发送请求给服务端返回的异步请求，一定是可以获取到结果的，只不过是执行时间的问题。客户端第一次查看异步请求可能还没完成，第二次再查看异步请求可能就完成了。那么如果不先处理异常情况，而是先发送请求，然后在服务端进行错误处理，这样是否可能呢？下面代码段第二部分服务端处理异常情况，返回异常的异步请求(完成，但失败)，这样客户端还是有机会进行错误重试的。虽然这种方案可行，但因为要指定发送请求的目标节点，对于前面的两种异常情况(没有分区或分区没有主副本)都没有目标节点，所以这种方案是不可取的。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 8]

正常情况把请求发送出去，异常情况不会发送请求，两者在异步请求的使用上还有以下不同点。

*  只有异常情况才会调用外层的循环。正常情况下发送了请求，但还没有收到结果，不会执行外层的循环，只会执行内层的循环。如果正常情况也要执行外层循环，就会发送多次请求了，而实际上只需要发送一次请求，然后等待这个请求完成。
 *  异常情况下调用cli.ent.poll(future)会立即返回，然后会隔段时间重试，直到可以发送请求为止。正常情况下有网络通信，调用cli.ent.poll(future)会阻塞住。只有客户端轮询到结果，异步请求完成了，才会执行后面的语句。

如图4-31所示，异步请求的“是否完成”和“是否成功”两个变量决定了怎么执行两层while循环。

(1)完成但不成功，说明出现了异常情况，执行外层循环。
    (2)未完成，说明没有异常情况，而且发送了请求，但还没有收到响应结果，执行内层循环。
    (3)完成而且成功，说明在步骤(2)的基础上收到了响应结果，结束外层循环。

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

从“获取分区偏移量”的示例上来看，异步请求中与结果相关的方法和变量如下。

*  complete(value)方法。解析客户端响应结果，井设置到异步请求的value变量中。
 *  rai…se(excepti…on)方法。出现异常情况，设置异常信息以及i…sDone=true。
 *  value变量。获取complete()方法设置的值，在异步请求完成后获取结果。0i…sDone变量。判断异步请求是否完成，成功和失败都算完成，都会设置1.sDone=true。
 *  succeeded()方法。异步请求是否成功，只有i…sDone=true而且没有异常，才算成功。

按照LIST\_OFFSETS的示例，我们可以模仿拉取请求使用“组合加适配器”模式的用法。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 10]

注意：使用组合模式返回的异步请求，必须在适配器的回调方法中调用异步请求的complete()方法，最后才可以获取异步请求的结果。上面代码中出现的4个异步请求对象都指组合返回的异步请求，而不是发送请求返回的异步请求。

总结一下异步请求使用监听器、组合加适配器模式相关的方法返回值。

(1)客户端调用send()发送请求，返回异步请求。
    (2)在异步请求上添加一个监昕器，没有返回值。
    (3)在异步请求上组合一个适配器，返回新的异步请求。
    (4)在异步请求上调用value()方法，返回异步请求的结果。

组合加适配器模式返回的还是一个异步请求，所以还可以在组合的返回结果上再添加一个监昕器，甚至可以再组合新的异步请求。另外，异步请求可以管理多个监昕器，可以在异步请求上添加多个监听器。下面两段代码模拟了为组合模式的异步请求添加监听器，以及为普通模式添加多个监昕器：  
![在这里插入图片描述][20210408221249275.png]

除了组合模式，异步请求还有更加高级的模式，比如链接模式会将两个异步请求链接在一起。

1.  异步请求的链式调用

异步请求可以通过chain()方法将另一个异步请求链接起来，它和组合模式的对比如下。

*  组合模式和链接模式都会为当前异步请求添加一个监昕器。
 *  组合模式会创建一个新异步请求，链接模式则传入一个已有的异步请求。
 *  组合模式返回新异步请求，链接模式返回当前异步请求，不是传入的已有异步请求。

相关代码如下：

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

如图4-32所示，消费者加入消费组结合消费者网络客户端的轮询，具体的执行步骤如下。

(1)消费者在轮询时为了确保分配到分区，会向协调者发送“加入组请求”。
    (2)“加入组请求”通过组合模式定义加入组的响应处理器，并返回异步请求。
    (3)请求被发送后，客户端会轮询，并等待服务端返回加入组的响应结果。
    (4)服务端返回加入组的响应结果，步骤(2)的响应处理器被触发。
    (5)加入组的响应处理器回调方法中，会再次向协调者发送“同步组请求”。
    (6)“同步组请求”也会定义同步组的响应处理器，也返回异步请求。
    (7)为“同步组请求”的异步请求链接上“加入组请求”的异步请求。
    (8)链接模式会为同步组的异步请求添加一个监昕器。
    (9)客户端会轮询并等待服务端返回同步组的响应结果。
    (10)服务端返回同步组的响应结果，步骤(6)的响应处理器被触发。
    (11)从同步组的客户归应结果中解析出字节数组，完成同步组异步请求。
    (12)同步组异步请求完成了，将字节数组设置到同步组异步请求的结果值中。
    (13)同步组异步请求在步骤(8)中有监昕器，触发调用监听器的回调方法。
    (14)监听器调用加入组异步请求的complete()方法，完成加入组异步请求。
    (15)同步组异步请求的结果值会作为加入组异步请求的结果，两者的类型也一样。
    (16)步骤(1)在发送“加入组请求”后，现在终于可以读取到加入组异步请求的结果值了。

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

上面大致的流程是：发送“加入组请求”→返回加入组的异步请求→完成“加入组请求”→发送“同步组请求”→返回同步组的异步请求→完成“同步组请求”→完成同步组的异步请求→完成加入组的异步请求→获取加入组异步请求的结果。注意这里第三步：完成“加入组请求”，表示收到加入组的客户端响应，但是这时因为还没有调用加入组异步请求的完成方法，并没有完成加入组的异步请求。加入组的异步请求只在倒数第二步，通过同步组异步请求的链接方法调用才完成。总结一下这些步骤的特点。

*  发送“加入组请求”先于发送“同步组请求”，但完成同步组异步请求先于完成加入组异步请求。
 *  组合方法会创建新的异步请求创建加入组的异步请求也先于创建同步组的异步请求。
 *  同步组异步请求的结果和加入组异步请求的结果不仅数据一样，类型也一样。
 *  如果发送“加入组请求”没有生成客户端响应结果，就不会发送“同步组请求”。
 *  客户端轮询只会调用一次响应处理器，这里可以完成异步请求比如完成同步组异步请求。
 *  加入组的响应处理器没有完成加入组的异步请求，只能通过链接方法由其他对象来间接完成。

这一节分析了异步请求的调用流程以及多种模式的使用。消费者有很多发送请求的方式都采用异步请求，后面如果出现和异步请求相关的调用流程，就不会再详细分析了。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/0cc498ffd10144bcbfd65635d9a73f84.png
[20210408220551741.png]: /images/20221022/2276204a54f74b73bd54f195751e4a43.png
[20210408220603803.png]: /images/20221022/86b9a1f7ee7642cba5bf09df35517686.png
[20210408220616349.png]: /images/20221022/4bf67614ccb447c79ba9280c7c970bcb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/85663e68f74845f28cbf7642d6fd94da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/37a8c6cce69542ec8069f167154bb9fd.png
[20210408220743398.png]: /images/20221022/16fa74daae8e4194b0fb739e4a51e70f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/7fc2ea513868471791d34f0a48ab5abc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/62d39fc135c244619b221ad25d5eed0c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/6d464b016e0a4338a418c36a34e94a39.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]: /images/20221022/080d9bdc5961486ea162f7d0806091ac.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]: /images/20221022/6d94a47098624d71a10ccd132abdabd6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 8]: /images/20221022/2c343870c47e4eaab116ae6c93581252.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 9]: /images/20221022/a54757b5b8404b57b40c2e31b2baa811.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 10]: /images/20221022/b8d441fe7ad241dc97e217f9db2aa3cc.png
[20210408221249275.png]: /images/20221022/b9caed29dc3e420683382b8990d57db4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 11]: /images/20221022/4817b8b6a303452d9dd31fa89499ab52.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 12]: /images/20221022/39661f90eaad4b59b36a4a3d5748047d.png
[20210408221336401.png]: /images/20221022/5595c84fdfa54192b9a79bdffbfd670f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 13]: /images/20221022/f39ef3812eea4744b552e6ef155d88fc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 14]: /images/20221022/605f7da783e24a69abf661102a379500.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 15]: /images/20221022/039370821c254f04ae801f6048f69935.png