如何根据MySQL负载动态调整Kafka消费者消费速率

不要用系统load average、整机CPU使用率作为MySQL负载的判断依据，这两个指标容易被服务器上其他无关进程干扰，误判率极高。生产环境通常采用多指标加权计算的方式得到负载值，覆盖MySQL运行的核心瓶颈点：

• 指标权重与计算规则（总分为100分）
  • 活跃连接使用率（权重30%）：取Threads_running状态值（即正在执行查询的非Sleep状态连接，不是总连接数），除以数据库配置的max_connections得到占比，该值超过70%就存在连接打满风险。
  • MySQL进程CPU使用率（权重25%）：仅统计MySQL进程自身占用的用户态CPU占比，排除其他进程的资源消耗，阈值参考75%。
  • 慢查询占比（权重25%）：固定采样间隔内的慢查询增量除以总查询增量，正常业务该值应低于1%，超过5%说明查询已经出现堆积，响应延迟明显升高。
  • 数据盘IO等待占比（权重20%）：MySQL数据所在磁盘的iowait值，超过30%说明磁盘IO成为瓶颈，即使CPU占用不高也属于高负载状态。
• 指标获取方式
  核心数据库状态值可以直接通过SQL查询获取，不需要额外部署监控组件：

  SHOW GLOBAL STATUS WHERE Variable_name IN ('Threads_running', 'Queries', 'Slow_queries', 'Max_used_connections');
  

  CPU、iowait指标可以直接读取服务器/proc目录下的系统状态值，云数据库可以直接调用云厂商的内置监控接口获取。
• 动态调整规则
  总负载分超过80分时触发降速，低于40分时逐步提速，每次调整的步长不要太大（比如休眠时长每次增减3-5s），避免消费速度跳变导致业务波动。

你当前采用的time.sleep()硬阻塞方案存在稳定性隐患：confluent_kafka的消费者心跳是在poll()调用时发送的，如果两次poll中间阻塞几十秒，很容易超过max.poll.interval.ms的默认阈值（30s），触发消费组重平衡，轻则重复消费，重则整个消费组卡住。以下是按落地成本从低到高排序的更优方案：

• 原生参数动态控速（改造成本最低）
  不需要在消费逻辑中加阻塞，只需要动态调整消费者的拉取配置即可：
  • 调整max.poll.records（单次poll最多拉取的消息条数）、fetch.max.bytes（单次poll拉取的最大字节数）两个参数，高负载时把值调小（比如从单次拉500条降到50条），低负载时调回原值。整个过程消费者不会阻塞，正常发送心跳，完全不会触发重平衡。
  • 多实例消费场景下，可以直接在Kafka broker端给对应消费组设置消费速率配额，从broker层面控制向消费组推送消息的速度，稳定性更高。
• 本地缓冲+令牌桶控速（适配复杂处理逻辑）
  将消息拉取和业务处理逻辑解耦：消费者线程poll到消息后直接写入本地有界队列（长度根据业务积压容忍度设置，比如1000），不做任何阻塞；单独起工作线程池负责MySQL查询逻辑，另外启动一个监控线程每10s检测一次MySQL负载，用令牌桶算法给工作线程发放处理许可——负载低时每秒发放100个许可，负载高时每秒发放10个，工作线程拿到许可才从队列取消息处理。这种方案下消费者线程永远不会阻塞，不会出现重平衡问题，控速精度也远高于硬sleep。
• 根源降载（效果最优）
  控速只是兜底方案，优先从根源降低MySQL压力：把单条查询攒批处理，比如100条同类型查询合并为一条IN查询，能直接把MySQL QPS降低一个数量级；同时给所有查询条件加匹配的索引，消灭慢查询，负载会出现明显下降，比任何控速方案的效果都好。

注意：无论采用哪种控速方案，必须保证两次poll()调用的间隔不超过配置的max.poll.interval.ms值，否则必然触发消费组重平衡。

内容的提问来源于stack exchange，提问作者DeadLock