RabbitMQ消息隊列異步解耦與業(yè)務(wù)削峰同步調(diào)用就像你打電話等對方接——對方不接你就一直卡著異步消息就像發(fā)微信——發(fā)完該干嘛干嘛對方有空了自然回你。一、消息隊列解決了什么問題在單體架構(gòu)時代所有功能揉在一個項目里方法之間直接調(diào)用簡單粗暴。但一旦系統(tǒng)變大問題就來了異步處理用戶注冊后要發(fā)郵件、發(fā)短信、發(fā)優(yōu)惠券……同步調(diào)用的話用戶得等半天體驗極差。丟到消息隊列里注冊接口秒回后續(xù)操作慢慢消費。應(yīng)用解耦訂單系統(tǒng)直接調(diào)用庫存系統(tǒng)庫存掛了訂單也跟著掛。中間加個隊列訂單只管發(fā)消息庫存恢復(fù)了繼續(xù)消費即可。流量削峰秒殺場景瞬間涌入10萬請求數(shù)據(jù)庫直接被干趴。隊列做個緩沖消費者按自己的節(jié)奏處理系統(tǒng)穩(wěn)如老狗。日志收集分布式系統(tǒng)中各服務(wù)把日志推到隊列由統(tǒng)一的日志服務(wù)消費存儲EFK/ELK的經(jīng)典套路。二、RabbitMQ核心概念RabbitMQ的消息流轉(zhuǎn)模型如下Producer → Exchange → (Binding) → Queue → Consumer 生產(chǎn)者 交換機(jī) 綁定 隊列 消費者Producer生產(chǎn)者產(chǎn)生消息的應(yīng)用程序Exchange交換機(jī)接收生產(chǎn)者發(fā)送的消息根據(jù)路由規(guī)則分發(fā)到隊列Queue隊列存放消息的緩沖區(qū)消息在這里排隊等消費Binding綁定交換機(jī)和隊列之間的關(guān)聯(lián)關(guān)系附帶路由鍵Consumer消費者從隊列中獲取消息并處理的應(yīng)用程序三、交換機(jī)四種類型RabbitMQ提供了四種Exchange類型理解清楚就知道消息怎么路由了。3.1 Direct直連最簡單的模式消息的路由鍵routing key和綁定的鍵完全匹配消息才會被投遞到對應(yīng)隊列。routing key order.create → 只匹配綁定 order.create 的隊列3.2 Fanout扇出廣播模式忽略路由鍵消息被投遞到與該交換機(jī)綁定的所有隊列。適合廣播通知場景。3.3 Topic主題支持通配符匹配靈活性最高*匹配一個單詞#匹配零個或多個單詞綁定鍵 order.* → 匹配 order.create、order.cancel不匹配 order.create.detail 綁定鍵 order.# → 匹配 order.create、order.create.detail 全都匹配3.4 Headers頭部不靠路由鍵而是根據(jù)消息頭headers中的鍵值對匹配。用的少了解即可。四、SpringBoot整合RabbitMQ4.1 引入依賴dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency4.2 yml配置spring:rabbitmq:host:127.0.0.1port:5672username:guestpassword:guest# 消息確認(rèn)機(jī)制publisher-confirm-type:correlated# 發(fā)布確認(rèn)publisher-returns:true# 消息返回listener:simple:acknowledge-mode:manual# 手動ACKprefetch:1# 每次拉取消息數(shù)4.3 隊列與交換機(jī)配置ConfigurationpublicclassRabbitMQConfig{// 隊列名稱publicstaticfinalStringEMAIL_QUEUEemail.queue;publicstaticfinalStringSMS_QUEUEsms.queue;publicstaticfinalStringORDER_EXCHANGEorder.exchange;publicstaticfinalStringORDER_ROUTING_KEYorder.notify;BeanpublicDirectExchangeorderExchange(){returnnewDirectExchange(ORDER_EXCHANGE,true,false);}BeanpublicQueueemailQueue(){returnnewQueue(EMAIL_QUEUE,true);}BeanpublicQueuesmsQueue(){returnnewQueue(SMS_QUEUE,true);}BeanpublicBindingemailBinding(QueueemailQueue,DirectExchangeorderExchange){returnBindingBuilder.bind(emailQueue).to(orderExchange).with(ORDER_ROUTING_KEY);}BeanpublicBindingsmsBinding(QueuesmsQueue,DirectExchangeorderExchange){returnBindingBuilder.bind(smsQueue).to(orderExchange).with(ORDER_ROUTING_KEY);}}五、發(fā)送消息RabbitTemplateServicepublicclassOrderService{AutowiredprivateRabbitTemplaterabbitTemplate;publicvoidcreateOrder(OrderDTOorderDTO){// 1. 保存訂單數(shù)據(jù)庫操作省略// ...// 2. 異步發(fā)送通知消息StringmsgJSON.toJSONString(orderDTO);rabbitTemplate.convertAndSend(RabbitMQConfig.ORDER_EXCHANGE,RabbitMQConfig.ORDER_ROUTING_KEY,msg);// 3. 直接返回不等郵件/短信發(fā)送完成return;}}六、接收消息RabbitListenerComponentpublicclassEmailConsumer{RabbitListener(queuesRabbitMQConfig.EMAIL_QUEUE)RabbitHandlerpublicvoidreceive(Stringmessage,Channelchannel,MessagemessageObj)throwsIOException{longdeliveryTagmessageObj.getMessageProperties().getDeliveryTag();try{OrderDTOorderJSON.parseObject(message,OrderDTO.class);// 發(fā)送郵件邏輯System.out.println(發(fā)送郵件到order.getEmail());// 手動確認(rèn)channel.basicAck(deliveryTag,false);}catch(Exceptione){// 消費失敗拒絕并重新入隊channel.basicNack(deliveryTag,false,true);}}}短信消費者結(jié)構(gòu)同理監(jiān)聽SMS_QUEUE即可。一個交換機(jī)綁定了兩個隊列同一條消息會同時投遞到郵件隊列和短信隊列實現(xiàn)并行處理。七、消息可靠性保障消息從生產(chǎn)到消費要經(jīng)過多個環(huán)節(jié)任何一個環(huán)節(jié)都可能丟消息。7.1 生產(chǎn)者確認(rèn)機(jī)制publisher-confirm-type:correlated# 異步確認(rèn)性能好rabbitTemplate.setConfirmCallback((correlationData,ack,cause)-{if(!ack){System.err.println(消息未到達(dá)Exchange原因cause);// 記錄日志重發(fā)等處理}});7.2 消費者手動ACK默認(rèn)是自動確認(rèn)auto消息一拿到就標(biāo)記消費成功但如果業(yè)務(wù)代碼報異常消息就丟了。改為手動確認(rèn)manual業(yè)務(wù)成功后調(diào)basicAck失敗調(diào)basicNack。八、死信隊列消息變成死信的三種情況消息被消費者rejectbasicReject/basicNack且不重新入隊消息TTL過期隊列或消息設(shè)置了過期時間隊列達(dá)到最大長度新消息被擠出去死信隊列的配置思路給正常隊列綁定一個死信交換機(jī)DLX消息變成死信后自動轉(zhuǎn)發(fā)到DLX再由DLX路由到死信隊列。BeanpublicQueuenormalQueue(){MapString,ObjectargsnewHashMap();args.put(x-message-ttl,60000);// 消息60秒過期args.put(x-dead-letter-exchange,dlx.exchange);args.put(x-dead-letter-routing-key,dlx.routing.key);returnnewQueue(normal.queue,true,false,false,args);}死信隊列常用于延遲任務(wù)消息過期→死信→消費、失敗消息重試、訂單超時取消等場景。九、常見問題與解決方案9.1 消息重復(fù)消費冪等性網(wǎng)絡(luò)抖動導(dǎo)致ACK沒及時到達(dá)RabbitMQ會重投消息消費者就重復(fù)處理了。解決方案業(yè)務(wù)唯一鍵校驗消費前查數(shù)據(jù)庫/Redis已處理則直接ACK跳過樂觀鎖update語句加where status 0條件Redis分布式鎖setnx保證同一消息只處理一次publicvoidreceive(Stringmessage){StringmsgIdextractMsgId(message);// Redis標(biāo)記已處理則跳過BooleanisNewredisTemplate.opsForValue().setIfAbsent(msg:processed:msgId,1,24,TimeUnit.HOURS);if(Boolean.FALSE.equals(isNew)){return;// 已處理過}// 正常消費邏輯}9.2 消息積壓處理消費速度跟不上生產(chǎn)速度隊列堆積越來越多的消息。應(yīng)對策略臨時擴(kuò)容消費者增加消費者實例數(shù)量批量消費一個消費者一次拉取多條消息處理消息轉(zhuǎn)存緊急將積壓消息轉(zhuǎn)存到另一個隊列后續(xù)慢慢消費根因排查消費者是不是有慢查詢是不是依賴的外部服務(wù)超時了RabbitMQ用好了就是系統(tǒng)穩(wěn)定性的護(hù)城河用不好就是給自己挖坑。把可靠性保障和冪等性設(shè)計到位消息隊列才能真正發(fā)揮價值。