mall整合RabbitMQ實現延遲消息

時間 2021-01-19

標籤 web spring api app 框架異步分佈式 ide spring-boot post 欄目 RabbitMQ 简体版

原文原文鏈接

本文主要講解mall整合RabbitMQ實現延遲消息的過程，以發送延遲消息取消超時訂單爲例。web

項目使用框架介紹

RabbitMQ

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

RabbitMQ的安裝和使用

1.安裝Erlang，下載地址：http://erlang.org/download/otpwin6421.3.exeapi

2.安裝RabbitMQ，下載地址：https://dl.bintray.com/rabbitmq/all/rabbitmq-server/3.7.14/rabbitmq-server-3.7.14.exeapp

3.安裝完成後，進入RabbitMQ安裝目錄下的sbin目錄框架

4.在地址欄輸入cmd並回車啓動命令行，而後輸入如下命令啓動管理功能：異步

rabbitmq-plugins enable rabbitmq_management

5.訪問地址查看是否安裝成功：http://localhost:15672/分佈式

6.輸入帳號密碼並登陸：guest guestide

7.建立賬號並設置其角色爲管理員：mall mallspring-boot

8.建立一個新的虛擬host爲：/mallpost

9.點擊mall用戶進入用戶配置頁面

10.給mall用戶配置該虛擬host的權限

11.至此，RabbitMQ的安裝和配置完成。

RabbitMQ的消息模型

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

Lombok

Lombok爲Java語言添加了很是有趣的附加功能，你能夠不用再爲實體類手寫getter,setter等方法，經過一個註解便可擁有。

注意：須要安裝idea的Lombok插件，並在項目中的pom文件中添加依賴。

業務場景說明

用於解決用戶下單之後，訂單超時如何取消訂單的問題。

用戶進行下單操做（會有鎖定商品庫存、使用優惠券、積分一系列的操做）；
生成訂單，獲取訂單的id；
獲取到設置的訂單超時時間（假設設置的爲60分鐘不支付取消訂單）；
按訂單超時時間發送一個延遲消息給RabbitMQ，讓它在訂單超時後觸發取消訂單的操做；
若是用戶沒有支付，進行取消訂單操做（釋放鎖定商品庫存、返還優惠券、返回積分一系列操做）。

整合RabbitMQ實現延遲消息

在pom.xml中添加相關依賴

<!--消息隊列相關依賴--><dependency>    <groupId>org.springframework.boot</groupId>    <artifactId>spring-boot-starter-amqp</artifactId></dependency><!--lombok依賴--><dependency>    <groupId>org.projectlombok</groupId>    <artifactId>lombok</artifactId>    <optional>true</optional></dependency>

修改SpringBoot配置文件

修改application.yml文件，在spring節點下添加RabbitMQ相關配置。

  rabbitmq:    host: localhost # rabbitmq的鏈接地址    port: 5672 # rabbitmq的鏈接端口號    virtual-host: /mall # rabbitmq的虛擬host    username: mall # rabbitmq的用戶名    password: mall # rabbitmq的密碼    publisher-confirms: true #若是對異步消息須要回調必須設置爲true

添加消息隊列的枚舉配置類QueueEnum

用於延遲消息隊列及處理取消訂單消息隊列的常量定義，包括交換機名稱、隊列名稱、路由鍵名稱。

package com.macro.mall.tiny.dto;
import lombok.Getter;
/**
* 消息隊列枚舉配置
* Created by macro on 2018/9/14.
*/
@Getter
public enum QueueEnum {
/**
* 消息通知隊列
*/
QUEUE_ORDER_CANCEL("mall.order.direct", "mall.order.cancel", "mall.order.cancel"),
/**
* 消息通知ttl隊列
*/
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 com.macro.mall.tiny.config;
import com.macro.mall.tiny.dto.QueueEnum;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 消息隊列配置
* Created by macro on 2018/9/14.
*/
@Configuration
public class RabbitMqConfig {
/**
* 訂單消息實際消費隊列所綁定的交換機
*/
@Bean
DirectExchange orderDirect() {
return (DirectExchange) ExchangeBuilder
.directExchange(QueueEnum.QUEUE_ORDER_CANCEL.getExchange())
.durable(true)
.build();
}
/**
* 訂單延遲隊列隊列所綁定的交換機
*/
@Bean
DirectExchange orderTtlDirect() {
return (DirectExchange) ExchangeBuilder
.directExchange(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange())
.durable(true)
.build();
}
/**
* 訂單實際消費隊列
*/
@Bean
public Queue orderQueue() {
return new Queue(QueueEnum.QUEUE_ORDER_CANCEL.getName());
}
/**
* 訂單延遲隊列（死信隊列）
*/
@Bean
public 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();
}
/**
* 將訂單隊列綁定到交換機
*/
@Bean
Binding orderBinding(DirectExchange orderDirect,Queue orderQueue){
return BindingBuilder
.bind(orderQueue)
.to(orderDirect)
.with(QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey());
}
/**
* 將訂單延遲隊列綁定到交換機
*/
@Bean
Binding 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 com.macro.mall.tiny.component;
import com.macro.mall.tiny.dto.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;
/**
* 取消訂單消息的發出者
* Created by macro on 2018/9/14.
*/
@Component
public class CancelOrderSender {
private static Logger LOGGER =LoggerFactory.getLogger(CancelOrderSender.class);
@Autowired
private AmqpTemplate amqpTemplate;
public void sendMessage(Long orderId,final long delayTimes){
//給延遲隊列發送消息
amqpTemplate.convertAndSend(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange(), QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey(), orderId, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//給消息設置延遲毫秒值
message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
return message;
}
});
LOGGER.info("send delay message orderId:{}",orderId);
}
}

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

用於從取消訂單的消息隊列（mall.order.cancel）裏接收消息。

package com.macro.mall.tiny.component;
import com.macro.mall.tiny.service.OmsPortalOrderService;
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;
/**
* 取消訂單消息的處理者
* Created by macro on 2018/9/14.
*/
@Component
@RabbitListener(queues = "mall.order.cancel")
public class CancelOrderReceiver {
private static Logger LOGGER =LoggerFactory.getLogger(CancelOrderReceiver.class);
@Autowired
private OmsPortalOrderService portalOrderService;
@RabbitHandler
public void handle(Long orderId){
LOGGER.info("receive delay message orderId:{}",orderId);
portalOrderService.cancelOrder(orderId);
}
}

添加OmsPortalOrderService接口

package com.macro.mall.tiny.service;
import com.macro.mall.tiny.common.api.CommonResult;
import com.macro.mall.tiny.dto.OrderParam;
import org.springframework.transaction.annotation.Transactional;
/**
* 前臺訂單管理Service
* Created by macro on 2018/8/30.
*/
public interface OmsPortalOrderService {
/**
* 根據提交信息生成訂單
*/
@Transactional
CommonResult generateOrder(OrderParam orderParam);
/**
* 取消單個超時訂單
*/
@Transactional
void cancelOrder(Long orderId);
}

添加OmsPortalOrderService的實現類OmsPortalOrderServiceImpl

package com.macro.mall.tiny.service.impl;
import com.macro.mall.tiny.common.api.CommonResult;
import com.macro.mall.tiny.component.CancelOrderSender;
import com.macro.mall.tiny.dto.OrderParam;
import com.macro.mall.tiny.service.OmsPortalOrderService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* 前臺訂單管理Service
* Created by macro on 2018/8/30.
*/
@Service
public class OmsPortalOrderServiceImpl implements OmsPortalOrderService {
private static Logger LOGGER = LoggerFactory.getLogger(OmsPortalOrderServiceImpl.class);
@Autowired
private CancelOrderSender cancelOrderSender;
@Override
public CommonResult generateOrder(OrderParam orderParam) {
//todo 執行一系類下單操做，具體參考mall項目
LOGGER.info("process generateOrder");
//下單完成後開啓一個延遲消息，用於當用戶沒有付款時取消訂單（orderId應該在下單後生成）
sendDelayMessageCancelOrder(11L);
return CommonResult.success(null, "下單成功");
}
@Override
public void cancelOrder(Long orderId) {
//todo 執行一系類取消訂單操做，具體參考mall項目
LOGGER.info("process cancelOrder orderId:{}",orderId);
}
private void sendDelayMessageCancelOrder(Long orderId) {
//獲取訂單超時時間，假設爲60分鐘
long delayTimes = 30 * 1000;
//發送延遲消息
cancelOrderSender.sendMessage(orderId, delayTimes);
}
}