图解kafka如何保证partition内部消息顺序（一）

短命女 2022-11-03 11:51 175阅读 0赞

> 原创不易，转载请注明出处  
> 注意：kafka版本v0.10

### 文章目录 ###

*   *   *  前言
         *  1.kafka保证消息顺序配置
         *  2.kafka生产者保证消息顺序原理
         *  3.总结

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

### 前言 ###

当面试官问我们kafka如何保证消息顺序的时候，熟悉kafka的同学可能都知道它是支持partition级别消息顺序性的，也就是保证在同一个partition内的消息是顺序的，接着面试官会让你说些实现消息顺序的配置参数，这个可能你会说出来，毕竟网上面经一堆，背背就ok了，但是一旦面试官问你它是怎样实现partition内消息顺序的时候，可能就不知道了，本文就是针对kafka保证消息顺序问题展开，从原理级别介绍它是如何保证消息顺序性的，本先介绍下kafka保证消息顺序是怎样配置的，接着就是介绍消息生产者发送消息的流程。我们在下一篇文章中会介绍broker与消费者端是怎样保证消息顺序的。

### 1.kafka保证消息顺序配置 ###

kafka producer：  
1. 失败重试配置不重试： retries=0这个默认就是0  
2. max.in.flight.requests.per.connection，这里解释下这个参数的意思，这个参数就是消息生产者将消息发送给broker了，但是broker还没有给回复的消息数量，也就是发送出去还没有响应的个数。它与上面那个重试参数配合使用，如果是不重试的话， max.in.flight.requests.per.connection配置多少都可以，默认是5个，但是你一旦重试的话，你这个参数必须要配置1，只允许一个请求一个响应的来，不能哗哗哗的将请求都发送出去了，这个时候有失败重试的顺序就会乱了。  
这里使用画图稍微解释下。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
比如说现在某个消息生产者是有5个消息，依次发送到broker 上。都还没有响应，这个时候  
in.flight.requests就是5  
如果这个时候2这个消息失败了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
这个时候就会对2这个消息进行重试，重新发送之后到broker就成了这个样子  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
如果你不重试的话，失败的消息的就丢了。但是能够保证没丢的消息整体顺序性。  
如果重试的话，这个max.in.flight.requests.per.connection 就得配置成1，这个时候发送消息之前就会判断待响应的数量是否大于等于1，如果是的话，就先不发送，等着响应回来并成功的时候再发送下一个消息包，如果失败的话，就重试。

kafka broker：这个不用配置，天然支持（这个解释原理的时候会介绍）  
kafka consumer：这个从broker拉取回来的消息是顺序的，然后拉取回来消息你自己处理的时候需要根据业务需求来考虑是否要保证顺序性

### 2.kafka生产者保证消息顺序原理 ###

先用一幅图来回顾下kafka发送消息的原理（原图：[链接][Link 1]）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]

这里解释下：

1.  首先，可以多个业务线程并发往内存缓冲区中写入消息，也可以单个线程往这个内存缓冲区  
    内存缓冲区作用就是进行batch，微批，将多个消息打成一个包，然后一次请求发送，这样能够减少网络请求的开销，增加吞吐量，但是你要控制好这个batch的大小，默认是16kb，如果这个batch设置的太大，然后你那个生产的频率还很低的话，这个时候你会发现延迟会很高，不过不用担心，kafka生产者还有个参数是linger.ms它管着就是不管你这个batch有没有满，只要创建时间超过了设置的时间，就会认为是准备好了。  
    消息到了会根据partition 分组，一个partition对应着一个队列，然后每个队列的元素是batch，一个batch 由若干个消息组成的，这消息在batch中也是有顺序的，毕竟就是内存buffer的追加写。
2.  业务线程写入内存缓冲区就不用管了，可以继续处理下一条消息的发送。这个时候kafka消息生产者有一个sender线程（注意是单线程）它负责的事情贼多，我们主要就是发送请求，接受响应，创建与broker的连接等等。  
    这里我们就介绍下与消息顺序有关的内容，  
    他会先找出那种准备好的node（这个node的就是一个broker实例，一个broker进程）怎样算是准备好呢？连接建立，有那种可以发送的batch，从每个partition中获取一个准备好的batch，注意是一个batch。这里这个准备好的意思是如果不是重试的batch，只要对应的node连接ok了，batch满了或者linger.ms时间到了就算是准备好了，重试的batch要过了重试间隔时间才可以，再就是判断这个某个node在InFlightRequests队列的请求数量，不超过max.in.flight.requests.per.connection 才可以。  
    找出准备好的node之后，接着就是从内存缓冲区中获取batch，每个partition中获取一个准备好的batch，接着就是封装请求，发送请求，发送请求的时候先去InFlightRequests中node对应队列中添加这个请求，接着再将这个请求放到对应的channel中，这个时候还不发送的，只是触发写事件，具体的发送是sender 线程的poll过程来执行的，使用的是java nio事件驱动，真正发送是感知到写事件的时候才会写到对应的输出流中。  
    再来分析一下这个过程，sender线程是单线程，按照程序的流程顺序执行，每个partition一个请求包中只会取一个batch，接着就是封装请求发送请求，这一轮算是完事了，接着进行下一轮，还是找出准备好的node，取出对应partition中第一个batch，封装请求，发送请求。
3.  接着再说说响应的问题，这个响应与请求的顺序是一样的，比如说我发送了请求1，请求2，先发送的请求1，再发送的请求2。你收的响应顺序绝对是响应1，响应2。  
    这里我给说下为什么  
    先是消息生产者，sender单线程保证先往tcp中写入请求1的流，再写入请求2的流，tcp协议是能够保证顺序性，到broker的时候，绝对是先给broker请求1，再给请求2，然后broker 处理的时候也是先处理完请求1，组织响应，放到一个队列中，然后写回来，这块涉及到broker保证消息顺序的机制，我们后面会解释。  
    响应来了之后会InFlightRequests 中找到对应的请求，如果是失败的，就会判断重试次数，重新放入这个内存缓冲区中，等待下次的发送。  
    其实可以看到发送消息是顺序的，出问题就出在失败重试这个环节，要么不重试，要么就是max.in.flight.requests.per.connection 这个参数设置成1，这样发送的时候会判断InFlightRequests 中对应node发送请求还未响应的数量，如果是1，说明已经发送出去一个请求还没有收到响应，收到响应之后，如果是成功InFlightRequests 的数量就是0了，接着会发送下一个请求，如果是失败的话，判断要不要重试，如果重试将要重试的batch 到内存缓冲区对应队列的头上去，下次发送消息会先获取这个重试的消息。如果这个一个不成功，就会一直重试，后面的batch就一个不能发送，这样有效的保证了消息在partition内部的顺序性。

### 3.总结 ###

本文介绍了kafka保证消息顺序的配置与消息生产者保证消息顺序的原理。  
消息生产者retries与max.in.flight.requests.per.connection 这两个参数尤为重要。  
要么不重试，丢数据保证顺序性，要么就是max.in.flight.requests.per.connection 设置成1，同时只能一个请求发送，响应回来成功后接着发送下个请求。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221024/4474822fcea04f7fb1328eebd15a0c1e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221024/7238232fa3644d64a907d99936fccd11.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: /images/20221024/408be53de3b64cb1b9704b498a67abf1.png
[Link 1]: https://www.processon.com/view/link/603b4a0ce0b34d42faa7c99f
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: /images/20221024/19c4ebf8f2054c86a4ac0a1507b2392a.png