mall整合RabbitMQ實現延遲消息

摘要

本文主要講解mall整合RabbitMQ實現延遲消息的過程，以發送延遲消息取消超時訂單爲例。RabbitMQ是一個被普遍使用的開源消息隊列。它是輕量級且易於部署的，它能支持多種消息協議。RabbitMQ能夠部署在分佈式和聯合配置中，以知足高規模、高可用性的需求。

項目使用框架介紹

RabbitMQ

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

RabbitMQ的安裝和使用

1. 安裝Erlang，下載地址：erlang.org/download/ot…

2. 安裝RabbitMQ，下載地址：dl.bintray.com/rabbitmq/al…

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

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

rabbitmq-plugins enable rabbitmq_management
複製代碼

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

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

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

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

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節點下添加Mongodb相關配置。

 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);
    }

}

複製代碼

添加OmsPortalOrderController定義接口

package com.macro.mall.tiny.controller;

import com.macro.mall.tiny.dto.OrderParam;
import com.macro.mall.tiny.service.OmsPortalOrderService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

/** * 訂單管理Controller * Created by macro on 2018/8/30. */
@Controller
@Api(tags = "OmsPortalOrderController", description = "訂單管理")
@RequestMapping("/order")
public class OmsPortalOrderController {
    @Autowired
    private OmsPortalOrderService portalOrderService;

    @ApiOperation("根據購物車信息生成訂單")
    @RequestMapping(value = "/generateOrder", method = RequestMethod.POST)
    @ResponseBody
    public Object generateOrder(@RequestBody OrderParam orderParam) {
        return portalOrderService.generateOrder(orderParam);
    }
}

複製代碼

進行接口測試

調用下單接口

注意：已經將延遲消息時間設置爲30秒

項目源碼地址

github.com/macrozheng/…

公衆號

mall項目全套學習教程連載中，關注公衆號第一時間獲取。