ToB企服应用市场:ToB评测及商务社交产业平台

标题: 高性能kafka消费 [打印本页]

---

作者: 嚴華    时间: 2024-10-3 20:08
标题: 高性能kafka消费
在进行spring-kafka消费的过程中，大部分人可能都碰到过kafka消息堆积的情况，尤其是大数据处置惩罚的场景，这时候就要想办法提高消费本领。
提高消费者的数量可以吗？我们知道，kafka的消费者个数是与kafka的分区数相干的，一个分区最多只能被一个消费者消费，也就是说，你在客户端开启的消费者个数即使凌驾了分区数，也不能提高消费本领，反而还占资源！
那既然一个分区只能被一个消费者消费，那我增长分区数不就可以多开消费者了？这在理论上确实是可行的，但是在运维层面会涉及到分区、数据的迁徙，而且这期间kafka是不可用的状态，成本太过高昂。
本方案从消费端的代码层面着手，看怎样计划实现一个高性能的kafka消费组件。

一、异步

默认情况下，spring-kafka采用的是同步消费模式，这种情况下消费本领被分区数限定，难以提拔。以是我们起首想到的异步消费：

异步消费，将消息的拉取与处置惩罚交由不同的线程处置惩罚，在硬件答应的前提下，可自由提高消费本领。
异步消费固然提高了消费本领，但是对于offset的提交却带来了挑战：因为默认情况下，offset是自动提交的，自动提交很可能导致异步未处置惩罚完的消息丢失；

二、手动提交

方案：

• consumer将offset加入本地列表进行维护；
  
• 异步worker回调更新offset状态；
  
• consumer遍历offset，探求可提交的最大offset；
  
• 提交最大offset到kafka，并清空本地已提交offset；
  

三、有序offset列表

从上图可知，为了进行手动提交，其中最关键的部分是手动维护的offset列表，为实现offset的提交，要达到的目标：

• 减少提交次数，每次只提交已完成的最大offset；
  
• offset列表已提交部分及时开释，保持弹性；
  
• 只管包管offset更新的有序性，提交时前面offset已完成，避免出现头部真空征象；
  

【101~109处置惩罚完之后，只有3次offset提交动作，109提交之后，109及之前的offset都会从offset列表中清除。】

但实际情况是：

• 异步worker执行进度无法预料，对offset回调更新无法包管顺序；
  
• 极有可能出现头部长时间无更新的真空征象；
  
• 非常情况下，由于头部offset不停不更新导致offset列表不停得不到开释从而导致内存溢出；
  

【103、105、106固然都处置惩罚完了，但是都还不能提交，因为这时候101和102都还未被处置惩罚。】

非常情况下会出现以下情况：

【随着时间的推移，低位的offset始终未被处置惩罚，这会导致后面全部已处置惩罚的offset都得不到提交，同时由于高位offset不断写入，低位的offset得不到清除，offset列表将急剧膨胀。】

问题：在异步执行的情况下怎样实现offset列表更新的有序性？

四、传统线程池

传统线程池调度策略：

• 当线程数< corePoolSize时，直接创建新线程执利用命；
  
• 当corePoolSize<线程数<maxPoolSize时，如果队列未满，使命入队等待；
  
• 当corePoolSize<线程数<maxPoolSize时，如果队列已满，创建新线程执利用命；
  
• 当线程数=maxPoolSize，且队列已满，则执行拒绝策略；
  

从传统线程池的调度策略来看，它是惰性的，这对我们当前场景来说有哪些问题？

• 在第3步中，后到的使命可能会比先到的使命先执行完；
  
• 在第4步中，拒绝策略会导致使命抛弃或由主线程插队；
  这两点都会导致无法包管使命的顺序执行！
  

五、饥渴线程池

目标：

• 确保先到的使命先执行；
  
• 包管使命等待时的公平性；
  
• 不能丢弃使命；
  

方案：

• 改变传统线程池调度策略；
  
• 利用公平队列包管使命等待的公平性；
  
• 完全摒除拒绝策略；
  
• put instead offer；
  

【对传统线程池关键部分进行改写】

新的调度策略：

• 当线程数< corePoolSize时，直接创建新线程执利用命；
  
• 当corePoolSize<线程数<maxPoolSize时，强制创建新线程执利用命；
  
• 当线程数=maxPoolSize，入队列等待；
  

六、微批处置惩罚

目标：

• 提高末了业务执行服从，最大化提拔性能；
  
• 低沉本地offset列表内存开销；
  
• 提高offset提交至kafka的服从；
  

方案：

• 基于micro-batch微批思想，consumer线程对批量消息进行切分、整合，将微批数据交给不同的worker线程处置惩罚；
  
• offset列表仅记录micro-batch中最大的offset；
  
• 仅提交最大offset到kafka；
  

【从kafka批量拉取200条消息，在内部进行分片、切分、合并，以50一组形成微批，提交给4个worker处置惩罚，同时，本地offset列表只记录每个微批中最大的offset：50、100、150、200，这样在提高末了业务执行服从、offset提交服从的同时，也极大地节流了内存开销。】

微批处置惩罚要求业务必须具有原子性，micro-batch这一批要么全部成功，要么全部失败！

七、last offset处置惩罚

问题：

• consumer线程在提交offset时，秉着‘有多少就提交多少’的思想，并不等待，以是在异步的情况下会出现最后一批消息的若干offset得不到提交的情况；
  
• 如果后续没有业务消息产生，那么这若干offset将永久得不到提交，如果此时发生rebalance，将发生重复消费；
  

方案：
基于sideCar模式，提供额外的监督器，动态监测本节点的offset列表，如若发现offset列表仍有未提交的offset，则会主动发送探针消息，驱动consumer进行poll及commit；

八、重写与重排

问题：

• 发生rebalance时，consumer实例可能会发生变化，用已过时的consumer提交offset会失败；
  
• 发生rebalance时，可能会发生重复消费，本地offset列表会重复添加；
  
• 发生rebalance时，由于micro-batch机制，拉取的offset可能比已有的全部offset要小，造成offset列表乱序；
  

方案：

• 加入offset列表时，检查consumer实例，如果发生变化，则对consumer进行重写；
  
• 滤重检查及offset列表重排序；
  

九、分区重分配

问题：

• 发生rebalance时，节点被分配的分区也可能发生变化，这时节点之前生存的未提交offset列表就成了脏数据，并常驻内存；
  
• 脏数据导致探针消息不停地被发送，但是本节点消费不到，引起网络滥发；
  

问题思索（怎么判断offset列表为脏数据？）：

• 判断offset列表大小不变为什么不可以？
  ——可能存在每次巡检时大小都不变的情况；
  
• 判断offset列表内容不变为什么不可以？
  ——可能线程池繁忙，新offset添加不停等待导致列表不变化；
  

方案：
每次添加offset时记录offset列表的更新时间戳，Offset Monitor定时检测，当经历了一个rebalance周期后，如果时间戳仍未更新，判断offset列表为脏数据，予以清除。

方案思索

• 时间戳长时间不更新意味着什么？
  —— 1.分区重分配；2.线程池繁忙，新offset添加不进去；
  
• 线程池繁忙的场景，为什么offset列表不会被误杀？
  —— 把时间线拉长，新offset加入等待一个rebalance周期后，一定会rebalance；
  

【关键代码】

这个案例给我们的启发：从空间维度转变为时间维度去判断，这种思索维度的转变以及思索尺度的放大，低沉了解决问题的复杂性，提供了更多的可能性。

十、极致性能

问题：

• 传统队列在读写时存在锁争用，在高并发场景下，线程不停地被挂起、恢复，上下文切换过程存在着很大的开销；
  
• 在x86架构下的CPU，在高并发场景下很容易形成伪共享(多个线程操纵不同的变量，但是变量处于雷同的缓存行，修改变量会使缓存行失效，甚至发生跨槽读取)；
  

【传统队列关键代码】

Disruptor：
Disruptor是一个高性能的队列，通过以下计划解决传统队列的锁争用及伪共享问题：

• 无锁计划：采用CAS无锁方式，包管线程安全，并提高服从;
  
• 环形数组：可避免频繁的垃圾接纳，同时数组对处置惩罚器的缓存机制更加友爱;
  
• 元素定位：环形数组长度为2^n，通过位运算，能快速取到元素;
  
• Cache padding：通过添加额外的无用信息，避免伪共享引发的性能问题。
  

【cas与cache padding】

【关键代码】

十一、总体架构

十二、结果测试

服务器：3 * 6核8G
消费逻辑：将14万CDC消息盘算更新后写入ES
消费设置：kafka 6分区、批量拉取500、32线程、2048队列
结果对比：

• 不利用组件：81分钟
  
• 亲缘线程池：6分41秒
  
• Disruptor：4分51秒
  
• 饥渴线程池：4分11秒
  
• 饥渴 + 微批：1分35秒
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4