前言

本系列博客基于B站谷粒商城，只作為本人學習總結使用。這里我會比較注重業(yè)務邏輯的編寫和相關配置的流程。有問題可以評論或者聯系我互相交流。原視頻地址谷粒商城雷豐陽版。本人git倉庫地址Draknessssw的谷粒商城

---

消息隊列使用場景

消息隊列主要有兩種形式的目的地

1. 隊列（queue）：點對點消息通信（point-to-point）

點對點式： ? 消息發(fā)送者發(fā)送消息，消息代理將其放入一個隊列中，消息接收者從隊列中獲 取消息內容，消息讀取后被移出隊列 ? 消息只有唯一的發(fā)送者和接受者，但并不是說只能有一個接收者

2. 主題（topic）：發(fā)布（publish）/訂閱（subscribe）消息通信
發(fā)布訂閱式： ? 發(fā)送者（發(fā)布者）發(fā)送消息到主題，多個接收者（訂閱者）監(jiān)聽（訂閱）這個 主題，那么就會在消息到達時同時收到消息

Docker安裝rabbitMQ

docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management 4369

25672 (Erlang發(fā)現&集群端口)
5672, 5671 (AMQP端口)
15672 (web管理后臺端口)
61613, 61614 (STOMP協(xié)議端口)
1883, 8883 (MQTT協(xié)議端口)

---

交換機的四種分發(fā)策略

1、樸實無華的交換機綁定隊列方式direct，交換機綁定一個至多個隊列，交換機發(fā)送消息的binding-key需要和對應的routing-key信息一致。

2、廣播模式，路由綁定隊列，發(fā)送消息，不需要對應某個routing-key，向任何一個binding-key發(fā)送消息都可以發(fā)送到全部隊列。

3、主題路由模式，需要指定binding-key的通配符，符合通配符規(guī)則的才向對應的隊列發(fā)送消息。

---

消息確認機制

發(fā)送端確認

定制rabbitTemplate

首先是消息轉json

@Beanpublic MessageConverter messageConverter() {return new Jackson2JsonMessageConverter();}

消息正確抵達回調方法和失敗回調方法

/*** 定制RabbitTemplate* 1、服務收到消息就會回調* 1、spring.rabbitmq.publisher-confirms: true* 2、設置確認回調* 2、消息正確抵達隊列就會進行回調* 1、spring.rabbitmq.publisher-returns: true* spring.rabbitmq.template.mandatory: true* 2、設置確認回調ReturnCallback** 3、消費端確認(保證每個消息都被正確消費，此時才可以broker刪除這個消息)**/// @PostConstruct //MyRabbitConfig對象創(chuàng)建完成以后，執(zhí)行這個方法public void initRabbitTemplate() {/*** 1、只要消息抵達Broker就ack=true* correlationData：當前消息的唯一關聯數據(這個是消息的唯一id)* ack：消息是否成功收到* cause：失敗的原因*///設置確認回調rabbitTemplate.setConfirmCallback((correlationData,ack,cause) -> {System.out.println("confirm...correlationData["+correlationData+"]==>ack:["+ack+"]==>cause:["+cause+"]");});/*** 只要消息沒有投遞給指定的隊列，就觸發(fā)這個失敗回調* message：投遞失敗的消息詳細信息* replyCode：回復的狀態(tài)碼* replyText：回復的文本內容* exchange：當時這個消息發(fā)給哪個交換機* routingKey：當時這個消息用哪個路郵鍵*/rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey) -> {System.out.println("Fail Message["+message+"]==>replyCode["+replyCode+"]" +"==>replyText["+replyText+"]==>exchange["+exchange+"]==>routingKey["+routingKey+"]");});}

rabbitＭＱ配置

spring:rabbitmq:host: 192.168.75.129port: 5672# 虛擬主機virtual-host: /# 開啟發(fā)送端發(fā)送確認，無論是否到達broker都會觸發(fā)回調【發(fā)送端確認機制+本地事務表】publisher-confirm-type: correlated# 開啟發(fā)送端抵達隊列確認，消息未被隊列接收時觸發(fā)回調【發(fā)送端確認機制+本地事務表】publisher-returns: true# 消息在沒有被隊列接收時是否強行退回template:mandatory: true# 消費者手動確認模式，關閉自動確認，否則會消息丟失listener:simple:acknowledge-mode: manual

---

消費端可以自動確認消息的抵達
但是，可以設置消費端手動確認消息抵達

example

package com.xxxx.gulimall.order.listener;import com.rabbitmq.client.Channel; import com.xxxx.gulimall.order.entity.OrderEntity; import com.xxxx.gulimall.order.service.OrderService; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service;import java.io.IOException;/*** @Description: 定時關閉訂單***/@RabbitListener(queues = "order.release.order.queue") @Service public class OrderCloseListener {@Autowiredprivate OrderService orderService;@RabbitHandlerpublic void listener(OrderEntity orderEntity, Channel channel, Message message) throws IOException {System.out.println("收到過期的訂單信息，準備關閉訂單" + orderEntity.getOrderSn());try {orderService.closeOrder(orderEntity);channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);} catch (Exception e) {channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);}}}

---

消息丟失處理

try {//TODO 確保每個消息發(fā)送成功，給每個消息做好日志記錄，(給數據庫保存每一個詳細信息)保存每個消息的詳細信息rabbitTemplate.convertAndSend("order-event-exchange", "order.release.other", orderTo);} catch (Exception e) {//TODO 定期掃描數據庫，重新發(fā)送失敗的消息}

---

消息重復

---

消息積壓

總結

以上是生活随笔為你收集整理的2022谷粒商城学习笔记（二十二）rabbitＭＱ学习的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。