## RabbitMQ

RabbitMQ是一個被廣泛使用的開源消息隊列。它是輕量級且易于部署的，它能支持多種消息協議。RabbitMQ可以部署在分布式和聯合配置中，以滿足高規模、高可用性的需求。

RabbitMQ的安裝與部署

在這里就不在陳述了，如果需要的話請看這篇博客https://www.cnblogs.com/chy-op/p/9611124.html

RabbitMQ的消息模型

標志中文名英文名描述P生產者Producer消息的發送者，可以將消息發送到交換機C消費者Consumer消息的接收者，從隊列中獲取消息進行消費X交換機Exchange接收生產者發送的消息，并根據路由鍵發送給指定隊列Q隊列Queue存儲從交換機發來的消息type交換機類型typedirect表示直接根據路由鍵（orange/black）發送消息

半藏商城中應用消息隊列的場景

1、用于解決用戶下單以后，訂單超時如何取消訂單的問題。 - 用戶進行提交訂單操作（會有鎖定商品庫存等操作）； - 生成訂單，獲取訂單的id； - 獲取到設置的訂單超時時間（假設設置的為60分鐘不支付取消訂單）； - 按訂單超時時間發送一個延遲消息給RabbitMQ，讓它在訂單超時后觸發取消訂單的操作； - 如果用戶沒有支付，進行取消訂單操作（釋放鎖定商品庫存一系列操作）。
實現方法 - 需要一個訂單延遲消息隊列 以及一個取消訂單消息隊列， - 一旦有消息以延遲訂單設置的路由鍵發送過來，會轉發到訂單延遲消息隊列，并在此隊列保存一定時間，等到超時后會自動將消息發送到取消訂單消息消費隊列。
2、短信驗證碼以及郵箱驗證碼都采用消息隊列進行消費。 - 采用隊列，交換機，路由鍵進行消費。一條隊列，一個交換機，一個路由鍵就可以實現。

在pom.xml中添加相關依賴

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId> </dependency>

修改SpringBoot配置文件

#SpringBoot配置RabbitMqrabbitmq:host: localhost # rabbitmq的連接地址port: 5672 # rabbitmq的連接端口號virtual-host: /hanzoMall # rabbitmq的虛擬hostusername: hanzoMall # rabbitmq的用戶名password: hanzoMall # rabbitmq的密碼publisher-confirms: true #如果對異步消息需要回調必須設置為true

消息隊列的枚舉配置類QueueEnum

用于延遲消息隊列及處理取消訂單消息隊列的常量定義，包括交換機名稱、隊列名稱、路由鍵名稱。 以及短信發送消息隊列和郵箱驗證碼發送消息隊列的常量定義，包括交換機名稱、隊列名稱、路由鍵名稱。package ltd.hanzo.mall.common;import com.rabbitmq.client.AMQP; import lombok.Getter;/*** @Author 皓宇QAQ* @email 2469653218@qq.com* @Date 2020/5/23 21:00* @link https://github.com/Tianhaoy/hanzomall* @Description: 消息隊列枚舉配置*/ @Getter public enum QueueEnum {/*** 發送短信消息通知隊列*/QUEUE_SMS_SEND("mall.sms.direct", "mall.sms.send", "mall.sms.send"),/*** 發送郵件消息通知隊列*/QUEUE_EMAIL_SEND("mall.email.direct", "mall.email.send", "mall.email.send"),/*** 消息通知隊列* mall.order.direct（取消訂單消息隊列所綁定的交換機）:綁定的隊列為mall.order.cancel，一旦有消息以mall.order.cancel為路由鍵發過來，會發送到此隊列。*/QUEUE_ORDER_CANCEL("mall.order.direct", "mall.order.cancel", "mall.order.cancel"),/*** 消息通知ttl隊列* mall.order.direct.ttl（訂單延遲消息隊列所綁定的交換機）:綁定的隊列為mall.order.cancel.ttl，一旦有消息以mall.order.cancel.ttl為路由鍵發送過來，會轉發到此隊列，并在此隊列保存一定時間，等到超時后會自動將消息發送到mall.order.cancel（取消訂單消息消費隊列）。*/QUEUE_TTL_ORDER_CANCEL("mall.order.direct.ttl", "mall.order.cancel.ttl", "mall.order.cancel.ttl");/*** 交換機名稱*/private String exchange;/*** 隊列名稱*/private String name;/*** 路由鍵*/private String routeKey;QueueEnum(String exchange, String name, String routeKey) {this.exchange = exchange;this.name = name;this.routeKey = routeKey;} }

添加RabbitMq的配置

用于配置交換機、隊列及隊列與交換機的綁定關系。package ltd.hanzo.mall.config;import ltd.hanzo.mall.common.QueueEnum; import org.springframework.amqp.core.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;/*** @Author 皓宇QAQ* @email 2469653218@qq.com* @Date 2020/5/23 21:04* @link https://github.com/Tianhaoy/hanzomall* @Description: 消息隊列配置*/ @Configuration public class RabbitMqConfig {/*** 1.0發送短信消息通知隊列所綁定的->交換機*/@BeanDirectExchange sendSmsDirect() {return (DirectExchange) ExchangeBuilder.directExchange(QueueEnum.QUEUE_SMS_SEND.getExchange()).durable(true).build();}/*** 1.0發送短信的->消費隊列*/@Beanpublic Queue sendSmsQueue() {return new Queue(QueueEnum.QUEUE_SMS_SEND.getName());}/*** 1.0將發送短信 隊列綁定到->交換機*/@BeanBinding sendSmsBinding(DirectExchange sendSmsDirect, Queue sendSmsQueue){return BindingBuilder.bind(sendSmsQueue).to(sendSmsDirect).with(QueueEnum.QUEUE_SMS_SEND.getRouteKey());}/*** 2.0發送郵件消息通知隊列所綁定的->交換機*/@BeanDirectExchange sendEmailDirect() {return (DirectExchange) ExchangeBuilder.directExchange(QueueEnum.QUEUE_EMAIL_SEND.getExchange()).durable(true).build();}/*** 2.0發送郵件的->消費隊列*/@Beanpublic Queue sendEmailQueue() {return new Queue(QueueEnum.QUEUE_EMAIL_SEND.getName());}/*** 2.0將發送郵件 隊列綁定到->交換機*/@BeanBinding sendEmailBinding(DirectExchange sendEmailDirect, Queue sendEmailQueue){return BindingBuilder.bind(sendEmailQueue).to(sendEmailDirect).with(QueueEnum.QUEUE_EMAIL_SEND.getRouteKey());}/*** 3.0訂單消息實際消費隊列所綁定的->交換機*/@BeanDirectExchange orderDirect() {return (DirectExchange) ExchangeBuilder.directExchange(QueueEnum.QUEUE_ORDER_CANCEL.getExchange()).durable(true).build();}/*** 3.0訂單實際消費隊列*/@Beanpublic Queue orderQueue() {return new Queue(QueueEnum.QUEUE_ORDER_CANCEL.getName());}/*** 3.0將訂單隊列綁定到交換機*/@BeanBinding orderBinding(DirectExchange orderDirect,Queue orderQueue){return BindingBuilder.bind(orderQueue).to(orderDirect).with(QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey());}/*** 4.0訂單延遲隊列隊列所綁定的->交換機*/@BeanDirectExchange orderTtlDirect() {return (DirectExchange) ExchangeBuilder.directExchange(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange()).durable(true).build();}/*** 4.0訂單延遲隊列（死信隊列）*/@Beanpublic Queue orderTtlQueue() {return QueueBuilder.durable(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getName()).withArgument("x-dead-letter-exchange", QueueEnum.QUEUE_ORDER_CANCEL.getExchange())//到期后轉發的交換機.withArgument("x-dead-letter-routing-key", QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey())//到期后轉發的路由鍵.build();}/*** 4.0將訂單延遲隊列綁定到交換機*/@BeanBinding orderTtlBinding(DirectExchange orderTtlDirect,Queue orderTtlQueue){return BindingBuilder.bind(orderTtlQueue).to(orderTtlDirect).with(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey());} }

在RabbitMQ管理頁面可以看到以下交換機和隊列

交換機及隊列說明

• mall.order.direct（取消訂單消息隊列所綁定的交換機）:綁定的隊列為mall.order.cancel，一旦有消息以mall.order.cancel為路由鍵發過來，會發送到此隊列。
• mall.order.direct.ttl（訂單延遲消息隊列所綁定的交換機）:綁定的隊列為mall.order.cancel.ttl，一旦有消息以mall.order.cancel.ttl為路由鍵發送過來，會轉發到此隊列，并在此隊列保存一定時間，等到超時后會自動將消息發送到mall.order.cancel（取消訂單消息消費隊列）。

添加延遲消息的發送者CancelOrderSender

用于向訂單延遲消息隊列（mall.order.cancel.ttl）里發送消息。package ltd.hanzo.mall.component;import lombok.extern.slf4j.Slf4j; import ltd.hanzo.mall.common.QueueEnum; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;/*** @Author 皓宇QAQ* @Date 2020/6/6 17:24* @Description:取消訂單消息的發出者*/ @Component @Slf4j public class CancelOrderSender {@Autowiredprivate AmqpTemplate amqpTemplate;public void sendMessage(String orderNo,final long delayTimes){//給延遲隊列發送消息amqpTemplate.convertAndSend(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange(), QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey(), orderNo, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {//給消息設置延遲毫秒值message.getMessageProperties().setExpiration(String.valueOf(delayTimes));return message;}});log.info("send delay message orderNo:{}",orderNo);} }

添加取消訂單消息的接收者CancelOrderReceiver

用于從取消訂單的消息隊列（mall.order.cancel）里接收消息。package ltd.hanzo.mall.component;import lombok.extern.slf4j.Slf4j; import ltd.hanzo.mall.service.HanZoMallOrderService; import ltd.hanzo.mall.service.TaskService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; 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.Component;/*** @Author 皓宇QAQ* @Date 2020/6/6 17:25* @Description:取消訂單消息的處理者*/ @Component @RabbitListener(queues = "mall.order.cancel") @Slf4j public class CancelOrderReceiver {@Autowiredprivate HanZoMallOrderService hanZoMallOrderService;@Autowiredprivate TaskService taskService;@RabbitHandlerpublic void handle(String orderNo){log.info("receive delay message orderNo:{}",orderNo);hanZoMallOrderService.cancelOrder(orderNo);taskService.cancelOrderSendSimpleMail(orderNo);} }

添加HanZoMallOrderService接口

public interface HanZoMallOrderService {/*** 保存訂單** @param user* @param myShoppingCartItems* @return*/String saveOrder(HanZoMallUserVO user, List<HanZoMallShoppingCartItemVO> myShoppingCartItems);/*** 取消單個超時訂單*/@Transactionalvoid cancelOrder(String orderNo);}

添加HanZoMallOrderService實現類HanZoMallOrderServiceImpl

@Slf4j @Service public class HanZoMallOrderServiceImpl implements HanZoMallOrderService {@Resourceprivate HanZoMallOrderMapper hanZoMallOrderMapper;@Resourceprivate HanZoMallOrderItemMapper hanZoMallOrderItemMapper;@Resourceprivate HanZoMallShoppingCartItemMapper hanZoMallShoppingCartItemMapper;@Resourceprivate HanZoMallGoodsMapper hanZoMallGoodsMapper;@Autowiredprivate CancelOrderSender cancelOrderSender;@Override@Transactionalpublic String saveOrder(HanZoMallUserVO user, List<HanZoMallShoppingCartItemVO> myShoppingCartItems) {//todo 執行一系類下單操作，代碼在github中//下單完成后開啟一個延遲消息，用于當用戶沒有付款時取消訂單 sendDelayMessageCancelOrder(orderNo);//所有操作成功后，將訂單號返回，以供Controller方法跳轉到訂單詳情return orderNo; ｝@Overridepublic void cancelOrder(String orderNo) {HanZoMallOrder hanZoMallOrder = hanZoMallOrderMapper.selectByOrderNo(orderNo);if (hanZoMallOrder != null && hanZoMallOrder.getOrderStatus() == 0) {//超時取消訂單hanZoMallOrderMapper.closeOrder(Collections.singletonList(hanZoMallOrder.getOrderId()), HanZoMallOrderStatusEnum.ORDER_CLOSED_BY_EXPIRED.getOrderStatus());}}private void sendDelayMessageCancelOrder(String orderNo) {//獲取訂單超時時間，假設為60分鐘long delayTimes = 36 * 100000;//發送延遲消息cancelOrderSender.sendMessage(orderNo, delayTimes);}

添加OrderController定義接口

代碼在Github中>https://github.com/Tianhaoy/hanzomall

總結

到此為止，整合RabbitMQ實現延遲消息的相關流程就介紹完畢了，知識只有分享出來才有價值。如果有問題的話，可以在關于我的頁面，通過我的郵箱聯系我進行探討。

總結

以上是生活随笔為你收集整理的rabbitmq 取消消息_SpringBoot整合RabbitMQ实现延迟消息的全部內容，希望文章能夠幫你解決所遇到的問題。

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