深度解析kafka broker从连接建立到接收请求发送响应

你的名字 2023-01-03 14:11 138阅读 0赞

> 原创不易，转载请注明出处  
> 阅读本文需要有java nio网络编程的基础

kafka 网络模块相关文章

1.《[深度解析kafka broker网络模型运行原理][kafka broker]》  
2.《[深度解析kafka broker网络组件初始化流程][kafka broker 1]》

### 文章目录 ###

*   *   *  前言
         *  1.建立连接
         *  2.接收请求
         *  3.处理请求
         *  4.发送响应
         *  总结

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

### 前言 ###

我们在前几篇kafka网络模块相关的文章中分别介绍了kafka broker 网络模型，运行流程，以及核心组件的初始化流程，本文我们将贴合实际场景出发，以消息生产者发送消息为例，看看acceptor是怎样处理新连接建立的，processor是怎样接收请求信息交给requestChannel的请求队列的，requestHandler又是怎样从请求队列取出消息，然后业务处理，封装响应，将响应放到响应队列的，processor又是怎样从自己的响应队列中获取响应，写回到对应的生产者客户端的。

### 1.建立连接 ###

之前的文章都介绍过，先连接请求是由acceptor组件处理并建立连接的。  
我们直接到Acceptor类的run方法里面看看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
首先是向selector注册OP\_ACCEPT事件的监听，接下来就是while 循环，不停的poll，阻塞时间是500ms。  
如果有ACCEPT 事件的发生，则会遍历这些事件，交给一个accept方法处理，我们来看下这个方法。processors(currentProcessor)这个东西就是取processors数组currentProcessor 位置的那个processor，就跟processors\[currentProcessor\]一个意思，不用太多的关注。  
关于这个currentProcessor 是怎么算出来的，下面有 `currentProcessor = (currentProcessor + 1) % processors.length`这么一行代码，仔细想想就是轮询的方式选择processor。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
首先调用serverSocketChannel的accept方法建立一个新连接拿到一个socketChannel，这都是jdk nio 的api，我们这里就不赘述。  
接着就是设置一些参数，设置非阻塞，不使用nagle算法，设置keepalive，发送缓冲区大小。这里有个注意的是上面connectionQuotas.inc(socketChannel.socket().getInetAddress) 这行，这个是控制一个ip 多少个连接的，默认是没限制。  
最后就是将这个新建立的连接交给processor处理了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
其实就是放到对应processor的newConnections 队列中了，然后唤醒下这个processor线程（此时processor线程可能在阻塞poll）

将新连接放到了processor 的newConnections 队列之后，我们看下processor它是怎么处理的。  
我们这里看下processor的run方法是怎样的处理逻辑  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
也是一个while循环不停的运行。  
configureNewConnections这个方法就是处理新连接的。我们先来看下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
我们看到就是不停的从newConnections队列中将socketChannel取出来，获取它的tcp四元组创建出来一个唯一的连接id。  
然后就是调用selector的register方法对这个channel进行注册，需要注意下这个selector是kafka自己的，里面是包装了jdk nio里面的那个selector，我们看下是怎样注册的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
三件事：

1.  将channel注册到jdk nio selector 上去，关注read事件。
2.  创建KafkaChannel，其实就是KafkaChannel将这个selectionKey包装了一下。需要注意这个key将KafkaChannel作为了一个附加参数，这里很重要，为啥呢？当selector 进行poll的时候，如果这个连接有请求过来，selector就会拿到这个selectionKey，然后在通过它的附加参数拿到这个KafkaChannel。
3.  将创建的这个channel放入channels这个map中缓存起来。

好了建立连接到这里就结束了。

### 2.接收请求 ###

连接建立好了，我们就当消息生产者发送一个消息过来，看看是怎样处理的，这里接着看processor对象的run方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
configureNewConnections这个方法我们上面看过了，就是处理新连接的，processNewResponses这个方法是处理新响应的，我们介绍响应的时候再看，现在我们看下这个poll方法  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]  
这里调用了kafka selector的poll方法，默认是堵塞300ms。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]  
上面那堆判断都先不要管，这个select方法里面其实就是调用jdk selector 的select方法，得到一个readyKeys，我们这个一定是可以有1个，毕竟消息生产者发送消息过来了。  
接着就会走pollSelectionKeys 这个方法，我们看下。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]  
这里就是遍历这些有事件的selectionKey，如果是读事件的话，就会走红框框出来的。  
有个while循环来不停的读，KafkaChannel的read方法进行读取，只要不是null就是一个请求，是null的话，说明这次没读完，就需要粘包处理了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]  
这次没读完，只能等着下次再poll过来的时候接着处理，如果读到一个完整的就调用addToStagedReceives 方法添加到暂存区里面，一个channel对应一个队列，添加到对应channel的队列里面。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]  
接着往下看，有个addToCompletedReceives方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 12]  
这个方法其实就是将stagedReceives 里面接收到的请求放到这个completedReceives集合中去，这里有个channel.isMute的这么一个判断需要注意下，是read事件的时候，才会从每个channel对应的队列中取出一个放入completedReceives 中。

好了这个poll方法我们这里就看完了，接着回到这个processor的run方法中  
接着看下processCompletedReceives 这个方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 13]  
其实就是遍历这个completedReceives集合每个元素，取出来然后转成request。将请求信息放入这RequestChannel的请求队列中去。  
下面还有个重要的就是selector.mute方法，移除对应channel对read事件的关注，这个说明了啥，下次循环poll的时候，如果不是关注这个read事件，就不会接收请求，也不会将stagedReceives 里面暂存的请求信息取出来。  
![在这里插入图片描述][20210109224615616.png]  
这个sendRequest就是将请求信息放入RequestChannel的请求队列中。

好了，获取请求信息部分的源码我们就看完了。

### 3.处理请求 ###

处理请求我们就要去requestHandler里面去看看了，直接看下它的run 方法  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 14]  
也是不断的while循环，从requestChannel中获取请求信息，然后交给这个apis的handle 方法处理，这个apis其实就是KafkaApis 。我们看下这个方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 15]  
可以看到就是区分请求的，看看是什么请求，然后找到对应的业务逻辑方法。  
这里我们是生产消息的请求，所以走handleProducerRequest方法，对应业务处理逻辑代码我们就不看了，我们只看看它的最后将响应信息放到那就可以了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 16]  
这个回调函数是在handleProducerRequest方法里的，这块就是处理完业务，然后封装响应信息放入对应processor的响应队列里。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 17]  
好了，这里就是requestHandler线程从RequestChannel的请求队列获取请求信息，然后处理业务，封装响应，放回到原来的那个processor对应的响应队列中的逻辑。接下来看看processor是怎样将响应信息拿出来，然后发送的。

### 4.发送响应 ###

我们再回到这个processor的run方法中  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 18]  
现在我们更关注的是processNewResponses 方法，poll 方法 与 processCompletedSends方法。  
先来看下这个processNewResponses  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 19]  
就是不停的从自己对应的那个响应队列中取出响应信息，然后调用sendResponse方法进行发送，这个发送不是真实的发送，只是暂存到kafka channel中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 20]  
调用selector 的send方法进行**暂存**，然后往这个inflightResponses 中添加这个响应信息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 21]  
其实就是找到对应的kafka channel 然后放到这个channel的send成员上暂存，然后关注channel 的write事件。  
到这 processNewResponses 方法作用很明朗了，就是将processor 自己对应的那个响应队列里面的那个响应信息取出来，然后放到对应kafka channel中暂存并关注这个write事件。

看到上面kafka channel的 setSend方法里面有个 send不是null的话就会抛出异常，那processor不停的取出来这个响应信息往对应channel里面放，会不会抛出异常？答案是不会的。  
为啥呢？这个其实要从接收消息那块说起，processor不停的读到新消息，然后就会将消息stagedReceives 里面，stagedReceives 是每个channel一个队列，然后就是遍历这个stagedReceives 每个channel中取出**一个请求信息出来**，而且还得是read事件的时候取出来放入completedReceives 集合中，接着后面又遍历这个completedReceives集合，将请求信息放入请求队列中，移除对应channel对read事件的关注，这个时候可以想想，就算是有新请求过来，你不关注这个read事件，其实就不会放入这个stagedReceives 里面，而stagedReceives里面剩余的也不会放入这个completedReceives 里面，最初的这批请求也是一个channel对应一个请求，这批请求响应没送出去之前，是不能处理新请求的，所以这个send就会一次设置。

接着看下poll方法处理这个写事件  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 22]  
这个就是真实的往channel里面写了。发送成功就会返回这个响应信息，添加到completedSends里面，但是发送没完成的话，就是null，不成功就涉及一个tcp 拆包的问题，我们来看下细节  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 23]  
可以看到，发送完成就会移除这个write事件关注，没完成的话，下次poll来还是write事件，还是接着往下发送，这就是拆包的处理。

最后来看看processor run 方法的processCompletedSends 方法，这个方法是对发送成功的响应信息进行处理。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 24]  
其实就是遍历这个completedSends 集合，然后从inflightResponses移除它，因为我们发送之前添加过。  
最后这个selector.unmute(send.destination) 很重要啊，其实就是找到对应channel ，让它关注read事件。。。  
关注完read事件之后，下次poll 又可以读新的请求信息了

### 总结 ###

1.  对kafka 网络线程模型 建立连接，接收请求，处理请求，发送响应整个链路的源码解析。
2.  各个组件是怎样协作的。
3.  对tcp 拆包，粘包的处理。
4.  单个channel 串行化的处理。

[kafka broker]: https://inetyoung.blog.csdn.net/article/details/112307717
[kafka broker 1]: https://inetyoung.blog.csdn.net/article/details/112357218
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221119/d97af194783e426a9807d94cbe76c3f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221119/bb581eefe811467d9b63195eb376ff5e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: /images/20221119/5c997c8d1d3b474ab1ec9c5133d1e60e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: /images/20221119/c583fea06fe1471583576b6a89832ed9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]: /images/20221119/65d215a6ab6c408babfd6fbf2db30277.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]: /images/20221119/ee2b51fa0d454b8381a761d8dac10631.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]: /images/20221119/f5a159002de44e16853f10d19729565f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]: /images/20221119/c3b1f67b4f934c16bbe789e160299b06.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]: /images/20221119/f739f24ca2424c5fad0c0d8fd208d02f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]: /images/20221119/2b1286d7c4034ab094cce539f92a38e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]: /images/20221119/dd831b6109bd4c3085e43922881a7179.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]: /images/20221119/a132818e60ea414aa1ad001eda8986f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 12]: /images/20221119/9219ff7df58a4f4aa33e4bb46faa6478.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 13]: /images/20221119/de98e8bee8304953918b411b6d994c92.png
[20210109224615616.png]: /images/20221119/68aad3f310b24c7e894b0ef112723094.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 14]: /images/20221119/8de17c0cc15448de8ae55c15794a88a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 15]: /images/20221119/f63574fcd92140c1b79e3e901424255c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 16]: /images/20221119/9cb57f93a92c4be081d557506417ec82.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 17]: /images/20221119/e08f5ae454e045e38cb44642098aadc4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 18]: /images/20221119/906c07cd41e64628ba6d44c46e9baf77.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 19]: /images/20221119/60baad938d524e21abbaa3d47a9ca874.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 20]: /images/20221119/b43d2aefa85e457a82e3532520c6d339.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 21]: /images/20221119/00c9275ebbf94e10b73fd8ed9cad62ec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 22]: /images/20221119/5024795e5ac348a7be2834d0735191f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 23]: /images/20221119/9e7944fee27045938d8664d9e216f9ff.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 24]: /images/20221119/580f5d2229e14c92904503a884cf54bb.png