做網站的網絡公司稅收優(yōu)惠,畫冊設計是什么,網站建設-好發(fā)信息網,網頁制作教程pdf半夜被電話叫醒#xff0c;消息積壓了200萬條#xff0c;消費者根本追不上。 這種場景搞過Kafka的應該都經歷過#xff0c;整理一下踩過的坑和解決方案。 坑一#xff1a;消息積壓 現象 監(jiān)控告警#xff1a;topic-order的lag超過100萬。 # 查看消費者lag kafka-consumer-g…半夜被電話叫醒消息積壓了200萬條消費者根本追不上。這種場景搞過Kafka的應該都經歷過整理一下踩過的坑和解決方案。坑一消息積壓現象監(jiān)控告警topic-order的lag超過100萬。# 查看消費者lagkafka-consumer-groups.sh --bootstrap-server localhost:9092--describe --group order-consumer GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG order-consumer topic-order0123456723456781111111order-consumer topic-order1123456823456791111111order-consumer topic-order2123456923456801111111三個分區(qū)每個積壓100多萬加起來300多萬。排查過程1. 先看生產速度# 查看topic的寫入速度kafka-run-class.sh kafka.tools.GetOffsetShell--broker-list localhost:9092--topic topic-order --time -1# 隔10秒再執(zhí)行一次算差值# 發(fā)現每秒寫入約5000條2. 再看消費速度消費者日志顯示處理一條消息要200ms算下來每秒只能處理5條。問題找到了消費太慢。解決方案方案一增加消費者實例Kafka的分區(qū)數決定了最大并行度。3個分區(qū)最多3個消費者并行。# 先增加分區(qū)注意分區(qū)只能增不能減kafka-topics.sh --bootstrap-server localhost:9092--alter --topic topic-order --partitions12然后部署12個消費者實例。方案二批量消費// 原來一條一條處理KafkaListener(topicstopic-order)publicvoidconsume(Stringmessage){processOrder(message);// 200ms}// 優(yōu)化后批量處理KafkaListener(topicstopic-order)publicvoidconsumeBatch(ListStringmessages){// 攢一批再處理減少IO次數batchProcessOrders(messages);// 批量寫庫}配置調整spring:kafka:consumer:max-poll-records:500# 一次拉取500條listener:type:batch# 批量模式方案三異步處理KafkaListener(topicstopic-order)publicvoidconsume(Stringmessage){// 扔到線程池異步處理executor.submit(()-processOrder(message));}但要注意異步處理需要手動管理offset提交不然可能丟消息。效果優(yōu)化后消費速度從5條/秒提升到3000條/秒積壓2小時內消化完?？佣G失現象業(yè)務反饋有訂單沒收到但生產端日志顯示發(fā)送成功了。排查1. 生產端配置props.put(acks,1);// 問題在這acks1表示leader收到就返回成功但如果leader掛了、follower還沒同步消息就丟了。2. 消費端配置props.put(enable.auto.commit,true);props.put(auto.commit.interval.ms,1000);自動提交offset如果消費處理到一半程序掛了offset已經提交了這條消息就丟了。解決方案生產端// acksall所有ISR副本都寫入才算成功props.put(acks,all);// 重試次數props.put(retries,3);// 開啟冪等性props.put(enable.idempotence,true);消費端// 關閉自動提交props.put(enable.auto.commit,false);// 手動提交KafkaListener(topicstopic-order)publicvoidconsume(ConsumerRecordString,Stringrecord,Acknowledgmentack){try{processOrder(record.value());ack.acknowledge();// 處理成功才提交}catch(Exceptione){// 處理失敗不提交會重新消費log.error(處理失敗,e);}}Broker端# 最小ISR副本數 min.insync.replicas2 # 不允許非ISR副本選舉為leader unclean.leader.election.enablefalse坑三重復消費現象同一條消息被處理了兩次導致訂單重復扣款。原因消費者處理完消息還沒來得及提交offset就掛了。重啟后從上次提交的offset開始消費這條消息又被消費一次。Kafka是at-least-once語義不保證exactly-once。解決方案業(yè)務冪等publicvoidprocessOrder(Stringmessage){OrderorderJSON.parseObject(message,Order.class);// 先查是否已處理過if(orderService.exists(order.getOrderId())){log.info(訂單已處理過跳過: {},order.getOrderId());return;}// 處理訂單orderService.process(order);}Redis去重publicvoidprocessOrder(Stringmessage){StringmsgIdextractMsgId(message);// Redis SETNX已存在返回falsebooleanisNewredis.setIfAbsent(kafka:processed:msgId,1,24,TimeUnit.HOURS);if(!isNew){log.info(消息已處理過: {},msgId);return;}// 處理業(yè)務doProcess(message);}數據庫唯一約束-- 用唯一約束兜底CREATEUNIQUEINDEXuk_order_idONorders(order_id);坑四消費者頻繁Rebalance現象日志里頻繁出現Revoking previously assigned partitions Rebalance triggered消費者不停地Rebalance效率極低。原因1. 心跳超時// 默認10秒沒心跳就認為消費者掛了session.timeout.ms10000如果處理一條消息超過10秒就會被踢出消費組。2. poll間隔太長// 默認5分鐘內必須調用pollmax.poll.interval.ms300000處理500條消息花了6分鐘超時了。解決方案// 增加session超時時間props.put(session.timeout.ms,30000);props.put(heartbeat.interval.ms,10000);// 增加poll間隔props.put(max.poll.interval.ms,600000);// 減少單次拉取數量props.put(max.poll.records,100);核心原則確保在max.poll.interval.ms內能處理完max.poll.records條消息?？游屙樞蛳M需求同一個用戶的操作必須按順序處理。問題默認情況下消息分散到不同分區(qū)不同分區(qū)的消費順序無法保證。解決方案指定分區(qū)key// 用userId作為key相同userId的消息會落到同一分區(qū)kafkaTemplate.send(topic-order,userId,message);單分區(qū)方案不推薦除非量很小// 只用一個分區(qū)保證全局順序kafkaTemplate.send(topic-order,0,null,message);注意事項同一分區(qū)內保證順序但重試可能打亂順序設置max.in.flight.requests.per.connection1保證嚴格順序props.put(max.in.flight.requests.per.connection,1);性能調優(yōu)參數生產者# 批量發(fā)送攢夠16K或等1ms就發(fā) batch.size16384 linger.ms1 # 發(fā)送緩沖區(qū) buffer.memory33554432 # 壓縮推薦lz4 compression.typelz4消費者# 單次拉取大小 fetch.min.bytes1 fetch.max.bytes52428800 fetch.max.wait.ms500 # 單次poll記錄數 max.poll.records500Broker# 日志保留 log.retention.hours168 log.retention.bytes1073741824 # 分區(qū)數根據消費者數量設置 num.partitions12 # 副本 default.replication.factor3 min.insync.replicas2監(jiān)控指標這幾個指標必須監(jiān)控指標含義報警閾值ConsumerLag消費延遲根據業(yè)務定MessagesInPerSec寫入速度突增報警BytesInPerSec流量接近帶寬報警UnderReplicatedPartitions副本不足的分區(qū)0報警OfflinePartitionsCount離線分區(qū)0報警集群運維我們的Kafka集群分布在兩個機房之前兩邊網絡不通很麻煩。后來用星空組網把兩個機房組到一個網絡里Kafka的跨機房復制配置簡單多了?？偨YKafka踩坑清單問題原因解決方案消息積壓消費慢加分區(qū)、批量消費、異步處理消息丟失acks配置不當acksall、手動提交重復消費at-least-once語義業(yè)務冪等、去重頻繁Rebalance超時配置不當調整超時參數順序問題多分區(qū)并行指定分區(qū)keyKafka本身很穩(wěn)定大多數問題都是配置和使用不當導致的。有Kafka相關問題歡迎評論區(qū)討論~