尚品汇总结九：RabbitMQ在项目的应用（面试专用）

左手的ㄟ右手 2023-10-14 13:53 11阅读 0赞

# 项目中的问题 #

# 1.搜索与商品服务的问题 #

商品服务修改了 商品的上架状态,商品就可以被搜索到.采用消息通知,商品服务修改完商品上架状态,发送消息 给 搜索服务,搜索服务消费消息,进行商品数据ES保存.下架也是一样.

## 2.订单服务取消订单问题 ##

延迟队里

保存订单之后 开始计时,时间到了,取消未支付的订单.

rabbitMQ的 延时消息.

## 3.分布式事务问题 ##

之前有一天 专门讲过分布式事务,都是概念.其中有一个 消息的最终数据一致性.

建议 每个人 回去把分布式事务的课件或者课程 看一遍.

场景:

支付----------订单----------库存 就是分布式事务场景.

## 4.秒杀的时候 ##

使用rabbitMQ进行消息通知、进行用户的排队。

# 消息队列解决什么问题 #

消息队列都解决了什么问题？

## 1、异步 ##

![12a5163627fb4f99a78bdf4ba4a10894.png][]

## 2、并行 ##

![244d8ca3f524443d97de0160fe9d7510.png][]

## 3、解耦 ##

![d50dd3c01d8b48d1bc8ea09b6bc0ee9d.png][]

## **4.排队 削峰 削去流量的峰值** ##

*  # **消息队列工具 RabbitMQ**  #

## **1 、常见MQ产品** ##

\- ActiveMQ：基于JMS（java协议）、消息类型比较少，2种 一对一，一对多的

\- RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好，消息类型多

\- RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会，使用也不少。

\- Kafka：分布式消息系统，高吞吐量，消息准确性不好

## **2 、RabbitMQ基础概念** ##

Broker：简单来说就是消息队列服务器实体

Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列

Queue：消息队列载体，每个消息都会被投入到一个或多个队列

Binding：绑定，它的作用就是把 exchange和 queue按照路由规则绑定起来

Routing Key：路由关键字， exchange根据这个关键字进行消息投递

vhost：虚拟主机，一个 broker里可以开设多个 vhost，用作不同用户的权限分离

producer：消息生产者，就是投递消息的程序

consumer：消息消费者，就是接受消息的程序

channel：消息通道，在客户端的每个连接里，可建立多个 channel，每个 channel代表一个会话任务

## **3、消息模型** ##

RabbitMQ提供了6种消息模型，但是第6种其实是RPC，并不是MQ，因此不予学习。那么也就剩下5种。

但是其实3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。

![9a786426641442c4a4bc7bf72d1da011.png][]

基本消息模型：生产者–>队列–>消费者

work消息模型：生产者–>队列–>多个消费者共同消费

订阅模型-Fanout：广播模式，将消息交给所有绑定到交换机的队列，每个消费者都会收到同一条消息

订阅模型-Direct：定向，把消息交给符合指定 rotingKey 的队列

订阅模型-Topic 主题模式：通配符，把消息交给符合routing pattern（路由模式） 的队列

*  # 消息不丢失 #

消息的准确性保证！！！！

消息的不丢失，在MQ角度考虑，一般有三种途径：

1.  **生产者不丢数据**

生产者 生产一个消息，准确的投递到交换机和队列中。

解决：rabbitMQ提供了 消息的发送确认，交换机应答和队列应答。

消息生产出来 向交互机中投递，如投递成功，会给true回执，失败了，会给false的回执。

交换机向队列路由消息的时候，如果消息没到达队列中，会触发队列应答，会执行对应应答的方法。

rabbitmq:
    
        host: 192.168.200.128
    
        port: 5672
    
        username: guest
    
        password: guest
    
        publisher-confirm-type: correlated #开启交换机应答
    
        publisher-returns: true   #队列应答
    
        listener:
    
          simple:
    
            acknowledge-mode: manual #默认情况下消息消费者是自动确认消息的，如果要手动确认消息则需要修改确认模式为manual
    
            prefetch: 1 # 消费者每次从队列获取的消息数量。此属性当不设置时为：轮询分发，设置为1为：公平分发

prefetch: 1 消费者每次从队列中消费消息的数量。

轮询分发：多个消费者的情况下，一人一次，不管消费方上次的消息有没有消费完，只要轮谁了就给谁。

公平分发：多个消费者的情况下，一人一次，如果轮到这个消费方了，但是它上一个消息还没有消费完，这个消息就给别人。只有上一个消息消费完了，才能消费下一个消息。

**2.MQ服务器不丢数据**

消息队列数据存在内存中，MQ挂了重启，内存容易释放，消息就没了。

提供了持久化，可以对 交换机、队列、消息 进行持久化。

**3.消费者不丢数据**

消费者在消费消息的过程中，可能会出现问题，消息还得存在。

手动签收消息，一旦消息消费过程中出问题了，可以拒绝签收，把消息转发到另外一个队列或者把消费异常的消息 做记录，等到问题解决了，再重新投递。

保证消息不丢失有两种实现方式：

1.  开启事务模式
2.  消息确认模式

说明：开启事务会大幅降低消息发送及接收效率，使用的相对较少。

在投递消息时开启事务支持，如果消息投递失败，则回滚事务，但是，很少有人这么干，因为这是同步操作，一条消息发送之后会使发送端阻塞，以等待RabbitMQ-Server的回应，之后才能继续发送下一条消息，生产者生产消息的吞吐量和性能都会大大降低。

因此我们生产环境一般都采取消息确认模式。

## 1、消息持久化 ##

如果希望RabbitMQ重启之后消息不丢失，那么需要对以下3种实体均配置持久化

Exchange

声明exchange时设置持久化（durable = true）并且不自动删除(autoDelete = false)

Queue

声明queue时设置持久化（durable = true）并且不自动删除(autoDelete = false)

message

发送消息时通过设置deliveryMode=2持久化消息

说明：

@Queue: 当所有消费客户端连接断开后，是否自动删除队列

true：删除 false：不删除

@Exchange：当所有绑定队列都不在使用时，是否自动删除交换器

true：删除 false：不删除

## 2、发送确认 ##

有时，业务处理成功，消息也发了，但是我们并不知道消息是否成功到达了rabbitmq，如果由于网络等原因导致业务成功而消息发送失败，那么发送方将出现不一致的问题，此时可以使用rabbitmq的发送确认功能，即要求rabbitmq显式告知我们消息是否已成功发送。

## 3、手动消费确认 ##

有时，消息被正确投递到消费方，但是消费方处理失败，那么便会出现消费方的不一致问题。比如订单已创建的消息发送到用户积分子系统中用于增加用户积分，但是积分消费方处理却失败了，用户就会问：我购买了东西为什么积分并没有增加呢？

要解决这个问题，需要引入消费方确认，即只有消息被成功处理之后才告知rabbitmq以ack，否则告知rabbitmq以nack

*  # 商品搜索上下架 #

## 1、service-product发送消息 ##

我在商品上架与商品添加时发送消息

商品上架

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <p style="margin-left:0;text-align:justify;"></p> <pre><code>实现类

@Override

@Transactional

public void onSale(Long skuId) {

&nbsp;&nbsp;&nbsp; // 更改销售状态

&nbsp;&nbsp;&nbsp; SkuInfo skuInfoUp = new SkuInfo();

&nbsp;&nbsp;&nbsp; skuInfoUp.setId(skuId);

&nbsp;&nbsp;&nbsp; skuInfoUp.setIsSale(1);

&nbsp;&nbsp;&nbsp; skuInfoMapper.updateById(skuInfoUp);

&nbsp;&nbsp;&nbsp; //商品上架

&nbsp;&nbsp;&nbsp; rabbitService.sendMessage(MqConst.EXCHANGE_DIRECT_GOODS, MqConst.ROUTING_GOODS_UPPER, skuId);

}</code></pre> <pre style="margin-left:0;text-align:justify;"></pre> </td> 
  </tr> 
 </tbody> 
</table>

商品下架

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <p style="margin-left:0;text-align:justify;"></p> <pre><code>实现类

@Override

@Transactional

public void cancelSale(Long skuId) {

&nbsp;&nbsp;&nbsp; // 更改销售状态

&nbsp;&nbsp;&nbsp; SkuInfo skuInfoUp = new SkuInfo();

&nbsp;&nbsp;&nbsp; skuInfoUp.setId(skuId);

&nbsp;&nbsp;&nbsp; skuInfoUp.setIsSale(0);

&nbsp;&nbsp;&nbsp; skuInfoMapper.updateById(skuInfoUp);

&nbsp;&nbsp;&nbsp; //商品下架

&nbsp;&nbsp;&nbsp; rabbitService.sendMessage(MqConst.EXCHANGE_DIRECT_GOODS, MqConst.ROUTING_GOODS_LOWER, skuId);

}</code></pre> <pre style="margin-left:0;text-align:justify;"></pre> <p style="margin-left:0;text-align:justify;"></p> </td> 
  </tr> 
 </tbody> 
</table>

## 2、service-list消费消息 ##

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <pre><code>package com.atguigu.gmall.list.receiver;

@Component

public class ListReceiver {

&nbsp;&nbsp;&nbsp; @Autowired

&nbsp;&nbsp;&nbsp; private SearchService searchService;

&nbsp;&nbsp;&nbsp; /**

&nbsp;&nbsp;&nbsp;&nbsp; * 商品上架

&nbsp;&nbsp;&nbsp;&nbsp; * @param skuId

&nbsp;&nbsp;&nbsp;&nbsp; * @throws IOException

&nbsp;&nbsp;&nbsp;&nbsp; */

&nbsp;&nbsp;&nbsp; @RabbitListener(bindings = @QueueBinding(

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; value = @Queue(value = MqConst.QUEUE_GOODS_UPPER, durable = "true"),

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_GOODS, type = ExchangeTypes.DIRECT, durable = "true"),

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; key = {MqConst.ROUTING_GOODS_UPPER}

&nbsp;&nbsp;&nbsp; ))

&nbsp;&nbsp;&nbsp; public void upperGoods(Long skuId, Message message, Channel channel) throws IOException {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (null != skuId) {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; searchService.upperGoods(skuId);

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

&nbsp;&nbsp;&nbsp; }

&nbsp;&nbsp;&nbsp; /**

&nbsp;&nbsp;&nbsp;&nbsp; * 商品下架

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; * @param skuId

&nbsp;&nbsp;&nbsp;&nbsp; */

&nbsp;&nbsp;&nbsp; @RabbitListener(bindings = @QueueBinding(

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; value = @Queue(value = MqConst.QUEUE_GOODS_LOWER, durable = "true"),

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_GOODS, type = ExchangeTypes.DIRECT, durable = "true"),

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; key = {MqConst.ROUTING_GOODS_LOWER}

&nbsp;&nbsp;&nbsp; ))

&nbsp;&nbsp;&nbsp; public void lowerGoods(Long skuId, Message message, Channel channel) throws IOException {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (null != skuId) {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; searchService.lowerGoods(skuId);

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

&nbsp;&nbsp;&nbsp; }

}</code></pre> <p></p> </td> 
  </tr> 
 </tbody> 
</table>

*  # 延迟队列关闭过期订单 #

延迟消息有两种实现方案：

**1.基于死信队列**

使用消息的存活时间，给队列设置参数，时间到了转发到另外一个队列，消费者真正消费的是 另一个队列。

**2.集成延迟插件**

延迟插件 中有个小型的库，延迟交换机把消息消息先存在小型的库中，时间到了再转发到队列。

区别：

死信的延迟消息：要求消息的时间一致，如果不一致，后面的消息出不来。

延迟插件的：没有这个要求。消息的时间可以不一致。

## 1 、基于死信实现延迟消息 ##

使用RabbitMQ来实现延迟消息必须先了解RabbitMQ的两个概念：消息的TTL和死信Exchange，通过这两者的组合来实现延迟队列

### 1.1、消息的TTL（Time To Live） ###

消息的TTL就是消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL。对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。

我们创建一个队列queue.temp，在Arguments 中添加x-message-ttl 为5000 （单位是毫秒），那所在压在这个队列的消息在5秒后会消失。

### 1.2、死信交换器 Dead Letter Exchanges ###

一个消息在满足如下条件下，会进死信路由，记住这里是路由而不是队列，一个路由可以对应很多队列。

（1） 一个消息被Consumer拒收了，并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其他消费者使用。

（2）上面的消息的TTL到了，消息过期了。

（3）队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上。

Dead Letter Exchange其实就是一种普通的exchange，和创建其他exchange没有两样。只是在某一个设置了Dead Letter Exchange的队列中有消息过期了，会自动触发消息的转发，发送到Dead Letter Exchange中去。

![f40e7f44cd2b4f0fade1c3726bd9b6f2.png][]

## 2 、基于延迟插件实现延迟消息 ##

### 2.1、插件安装 ###

1. 首先我们将刚下载下来的rabbitmq\_delayed\_message\_exchange-3.8.0.ez文件上传到RabbitMQ所在服务器，下载地址：https://www.rabbitmq.com/community-plugins.html

2. 切换到插件所在目录，执行 docker cp rabbitmq\_delayed\_message\_exchange-3.8.0.ez rabbitmq:/plugins 命令，将刚插件拷贝到容器内plugins目录下

3. 执行 docker exec -it rabbitmq /bin/bash 命令进入到容器内部，并 cd plugins 进入plugins目录

4. 执行 ls -l|grep delay 命令查看插件是否copy成功

5. 在容器内plugins目录下，执行 rabbitmq-plugins enable rabbitmq\_delayed\_message\_exchange 命令启用插件

6. exit命令退出RabbitMQ容器内部，然后执行 docker restart rabbitmq 命令重启RabbitMQ容器

### 2.2、代码实现 ###

配置队列

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <pre><code>package com.atguigu.gmall.mq.config;

@Configuration

public class DelayedMqConfig {

&nbsp;&nbsp;&nbsp; public static final String exchange_delay = "exchange.delay";

&nbsp;&nbsp;&nbsp; public static final String routing_delay = "routing.delay";

&nbsp;&nbsp;&nbsp; public static final String queue_delay_1 = "queue.delay.1";

&nbsp;&nbsp;&nbsp; /**

&nbsp;&nbsp;&nbsp;&nbsp; * 队列不要在RabbitListener上面做绑定，否则不会成功，如队列2，必须在此绑定

&nbsp;&nbsp;&nbsp;&nbsp; *

&nbsp;&nbsp;&nbsp;&nbsp; * @return

&nbsp;&nbsp;&nbsp;&nbsp; */

&nbsp;&nbsp;&nbsp; @Bean

&nbsp;&nbsp;&nbsp; public Queue delayQeue1() {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 第一个参数是创建的queue的名字，第二个参数是是否支持持久化

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return new Queue(queue_delay_1, true);

&nbsp;&nbsp;&nbsp; }

&nbsp;&nbsp;&nbsp; @Bean

&nbsp;&nbsp;&nbsp; public CustomExchange delayExchange() {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Map&lt;String, Object&gt; args = new HashMap&lt;String, Object&gt;();

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; args.put("x-delayed-type", "direct");

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return new CustomExchange(exchange_delay, "x-delayed-message", true, false, args);

&nbsp;&nbsp;&nbsp; }

&nbsp;&nbsp;&nbsp; @Bean

&nbsp;&nbsp;&nbsp; public Binding delayBbinding1() {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return BindingBuilder.bind(delayQeue1()).to(delayExchange()).with(routing_delay).noargs();

&nbsp;&nbsp;&nbsp; }

}</code></pre> <p style="margin-left:0;text-align:justify;"></p> </td> 
  </tr> 
 </tbody> 
</table>

发送消息

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <pre><code>@GetMapping("sendDelay")

public Result sendDelay() {

&nbsp;&nbsp; SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

&nbsp;&nbsp; this.rabbitTemplate.convertAndSend(DelayedMqConfig.exchange_delay, DelayedMqConfig.routing_delay, sdf.format(new Date()), new MessagePostProcessor() {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; @Override

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; public Message postProcessMessage(Message message) throws AmqpException {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; message.getMessageProperties().setDelay(10 * 1000);

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; System.out.println(sdf.format(new Date()) + " Delay sent.");

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return message;

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }

&nbsp;&nbsp; });

&nbsp;&nbsp; return Result.ok();

}</code></pre> <p style="margin-left:0;text-align:justify;"></p> </td> 
  </tr> 
 </tbody> 
</table>

接收消息

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <pre><code>package com.atguigu.gmall.mq.receiver;

@Component

@Configuration

public class DelayReceiver {

&nbsp;&nbsp;&nbsp; @RabbitListener(queues = DelayedMqConfig.queue_delay_1)

&nbsp;&nbsp;&nbsp; public void get(String msg) {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; System.out.println("Receive queue_delay_1: " + sdf.format(new Date()) + " Delay rece." + msg);

&nbsp;&nbsp;&nbsp; }

}</code></pre> <p style="margin-left:0;text-align:justify;"></p> </td> 
  </tr> 
 </tbody> 
</table>

## 3 、基于延迟插件实现取消订单 ##

rabbit-util模块延迟接口封装

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <pre><code>RabbitService
/**

&nbsp;* 发送延迟消息

&nbsp;* @param exchange 交换机

&nbsp;* @param routingKey 路由键

&nbsp;* @param message 消息

&nbsp;* @param delayTime 单位：秒

&nbsp;*/

public boolean sendDelayMessage(String exchange, String routingKey, Object message, int delayTime) {

&nbsp;&nbsp;&nbsp; GmallCorrelationData correlationData = new GmallCorrelationData();

&nbsp;&nbsp;&nbsp; String correlationId = UUID.randomUUID().toString();

&nbsp;&nbsp;&nbsp; correlationData.setId(correlationId);

&nbsp;&nbsp;&nbsp; correlationData.setMessage(message);

&nbsp;&nbsp;&nbsp; correlationData.setExchange(exchange);

&nbsp;&nbsp;&nbsp; correlationData.setRoutingKey(routingKey);

&nbsp;&nbsp;&nbsp; correlationData.setDelay(true);

&nbsp;&nbsp;&nbsp; correlationData.setDelayTime(delayTime);

&nbsp;&nbsp;&nbsp; redisTemplate.opsForValue().set(correlationId, JSON.toJSONString(correlationData), OBJECT_TIMEOUT, TimeUnit.MINUTES);

&nbsp;&nbsp;&nbsp; this.rabbitTemplate.convertAndSend(exchange, routingKey, message, new MessagePostProcessor() {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; @Override

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; public Message postProcessMessage(Message message) throws AmqpException {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; message.getMessageProperties().setDelay(delayTime*1000);

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return message;

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }

&nbsp;&nbsp;&nbsp; },correlationData);

&nbsp;&nbsp;&nbsp; return true;

}</code></pre> <pre style="margin-left:0;text-align:justify;"></pre> </td> 
  </tr> 
 </tbody> 
</table>

### 3.1、发送消息 ###

创建订单时，发送延迟消息

修改保存订单方法

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <pre><code>@Override

@Transactional

public Long saveOrderInfo(OrderInfo orderInfo) {

&nbsp;&nbsp;&nbsp; // orderInfo

&nbsp;&nbsp;&nbsp; // 总金额，订单状态，用户Id，第三方交易编号，创建时间，过期时间，进程状态

&nbsp;&nbsp;&nbsp; orderInfo.sumTotalAmount();

&nbsp;&nbsp;&nbsp; orderInfo.setOrderStatus(OrderStatus.UNPAID.name());

&nbsp;&nbsp;&nbsp; String outTradeNo = "ATGUIGU" + System.currentTimeMillis() + "" + new Random().nextInt(1000);

&nbsp;&nbsp;&nbsp; orderInfo.setOutTradeNo(outTradeNo);

&nbsp;&nbsp;&nbsp; orderInfo.setCreateTime(new Date());

&nbsp;&nbsp;&nbsp; // 定义为1天

&nbsp;&nbsp;&nbsp; Calendar calendar = Calendar.getInstance();

&nbsp;&nbsp;&nbsp; calendar.add(Calendar.DATE, 1);

&nbsp;&nbsp;&nbsp; orderInfo.setExpireTime(calendar.getTime());

&nbsp;&nbsp;&nbsp; orderInfo.setProcessStatus(ProcessStatus.UNPAID.name());

&nbsp;&nbsp;&nbsp; orderInfoMapper.insert(orderInfo);

&nbsp;&nbsp;&nbsp; StringBuffer tradeBody = new StringBuffer();

&nbsp;&nbsp;&nbsp; // 保存订单明细

&nbsp;&nbsp;&nbsp; List&lt;OrderDetail&gt; orderDetailList = orderInfo.getOrderDetailList();

&nbsp;&nbsp;&nbsp; for (OrderDetail orderDetail : orderDetailList) {

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; orderDetail.setId(null);

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; orderDetail.setOrderId(orderInfo.getId());

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; orderDetailMapper.insert(orderDetail);

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; tradeBody.append(orderDetail.getSkuName()).append(" ");

&nbsp;&nbsp;&nbsp; }

&nbsp;&nbsp;&nbsp; //更新支付描述

&nbsp;&nbsp;&nbsp; orderInfo.setTradeBody(tradeBody.toString());

&nbsp;&nbsp;&nbsp; orderInfoMapper.updateById(orderInfo);

&nbsp;&nbsp;&nbsp; //发送延迟队列，如果到过期时间了未支付，取消订单

&nbsp;rabbitService.sendDelayMessage(MqConst.EXCHANGE_DIRECT_ORDER_CANCEL, MqConst.ROUTING_ORDER_CANCEL, orderInfo.getId(), MqConst.DELAY_TIME);

&nbsp;&nbsp;&nbsp; // 返回

&nbsp;&nbsp;&nbsp; return orderInfo.getId();

}</code></pre> <p></p> </td> 
  </tr> 
 </tbody> 
</table>

### 3.2、接收消息 ###

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <pre style="margin-left:0;text-align:justify;"></pre> <pre><code>package com.atguigu.gmall.order.receiver;

@Component

public class OrderReceiver {

&nbsp;&nbsp;&nbsp; @Autowired

&nbsp;&nbsp;&nbsp; private OrderService orderService;

&nbsp;&nbsp;&nbsp; /**

&nbsp;&nbsp;&nbsp;&nbsp; * 取消订单消费者

&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;* 延迟队列，不能再这里做交换机与队列绑定
* 需要在配置类里去绑定死信交换机
&nbsp;&nbsp;&nbsp;&nbsp; * @param orderId
&nbsp;&nbsp;&nbsp;&nbsp; * @throws IOException
&nbsp;&nbsp;&nbsp;&nbsp; */
&nbsp;&nbsp;&nbsp; @RabbitListener(queues = MqConst.QUEUE_ORDER_CANCEL)
&nbsp;&nbsp;&nbsp; public void orderCancel(Long orderId, Message message, Channel channel) throws IOException {
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (null != orderId) {
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //防止重复消费
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; OrderInfo orderInfo = orderService.getById(orderId);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (null != orderInfo &amp;&amp; orderInfo.getOrderStatus().equals(ProcessStatus.UNPAID.getOrderStatus().name())) {
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; orderService.execExpiredOrder(orderId);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
&nbsp;&nbsp;&nbsp; }

}</code></pre> <p style="margin-left:0;text-align:justify;"></p> </td> 
  </tr> 
 </tbody> 
</table>

取消订单业务，取消订单要关闭支付交易

<table> 
 <tbody> 
  <tr> 
   <td style="border-color:#000000;vertical-align:top;width:426.1pt;"> <pre><code>@Override

public void execExpiredOrder(Long orderId) {

&nbsp;&nbsp;&nbsp; // orderInfo

&nbsp;&nbsp;&nbsp; updateOrderStatus(orderId, ProcessStatus.CLOSED);

&nbsp;&nbsp;&nbsp; // paymentInfo

&nbsp;&nbsp;&nbsp; //paymentFeignClient.closePayment(orderId);

&nbsp;&nbsp;&nbsp; //发送取消交易的消息

&nbsp;&nbsp;&nbsp; rabbitService.sendMessage(MqConst.EXCHANGE_DIRECT_PAYMENT_CLOSE, MqConst.ROUTING_PAYMENT_CLOSE, orderId);

}</code></pre> <p></p> </td> 
  </tr> 
 </tbody> 
</table>

关闭交易消息消费者

package com.atguigu.gmall.payment.receiver;
    
    @Component
    
    public class PaymentReceiver {
    
        @Autowired
    
        private PaymentService paymentService;
    
        /**
    
         * 取消交易
    
         * @param orderId
    
         * @throws IOException
    
         */
    
        @RabbitListener(bindings = @QueueBinding(
    
                value = @Queue(value = MqConst.QUEUE_PAYMENT_CLOSE, durable = "true"),
    
                exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_PAYMENT_CLOSE),
    
                key = {MqConst.ROUTING_PAYMENT_CLOSE}
    
        ))
    
        public void closePayment(Long orderId) throws IOException {
    
            if (null != orderId) {
    
                paymentService.closePayment(orderId);
    
            }
    
        }
    
    }

更改支付日志表，状态为关闭交易

@Override
    
    public void closePayment(Long orderId) {
    
        QueryWrapper<PaymentInfo> queryWrapper = new QueryWrapper<>();
    
        queryWrapper.eq("order_id", orderId);
    
        PaymentInfo paymentInfoUp = new PaymentInfo();
    
        paymentInfoUp.setPaymentStatus(PaymentStatus.ClOSED.name());
    
        paymentInfoMapper.update(paymentInfoUp, queryWrapper);
    
        //关闭交易
    
         alipayService.closePay(orderId);
    }
    支付宝支付AlipayServiceImpl实现类
    /***
    
     * 关闭交易
    
     * @param orderId
    
     * @return
    
     */
    
    @Override
    
    public Boolean closePay(Long orderId) {
    
        OrderInfo orderInfo = orderFeignClient.getOrderInfo(orderId);
    
    
    
        //AlipayClient alipayClient = new DefaultAlipayClient("https://openapi.alipay.com/gateway.do","app_id","your private_key","json","GBK","alipay_public_key","RSA2");
    
        AlipayTradeCloseRequest request = new AlipayTradeCloseRequest();
    
        HashMap<String, Object> map = new HashMap<>();
    
        map.put("trade_no", "");
    
        map.put("out_trade_no", orderInfo.getOutTradeNo());
    
        map.put("operator_id", "YX01");
    
    
    
        request.setBizContent(JSON.toJSONString(map));
    
        AlipayTradeCloseResponse response = null;
    
        try {
    
            response = alipayClient.execute(request);
    
        } catch (AlipayApiException e) {
    
            e.printStackTrace();
    
        }
    
        if(response.isSuccess()){
    
            log.info("调用成功");
    
            return true;
    
        }
    
        return false;
    
    }

*  # 项目中分布式事务的业务场景 #

![96f57d5d38b745bfaddd777526b0ac6d.png][]

*  # **RabbitMQ常见问题** #

## **1、使用RabbitMQ有什么好处？** ##

1）.解耦：系统A在代码中直接调用系统B和系统C的代码，如果将来D系统接入，系统A还需要修改代码，过于麻烦！

2）.异步：将消息写入消息队列，非必要的业务逻辑以异步的方式运行，加快响应速度

3）.削峰：并发量大的时候，所有的请求直接怼到数据库，造成数据库连接异常

## **2、消息顺序问题** ##

**场景：**比如支付操作，支付成功之后，会发送修改订单状态和扣减库存的消息，如果这两个消息同时发送，就不能保证完全按照顺序消费，有可能是先减库存了，后更改订单状态。

**解决方案：**同步执行，当一个消息执行完之后，再发布下一个消息。

## **3、如何保证RabbitMQ消息的可靠传输？** ##

消息不可靠的原因是因为消息丢失

**生产者丢失消息****:**

RabbitMQ提供transaction事务和confirm模式来确保生产者不丢消息；

Transaction事务机制就是说：发送消息前，开启事务（channel.txSelect()）,然后发送消息，如果发送过程中出现什么异常，事务就会回滚（channel.txRollback()）,如果发送成功则提交事务（channel.txCommit()），然而，这种方式有个缺点：吞吐量下降。

confirm模式用的居多：一旦channel进入confirm模式，所有在该信道上发布的消息都将会被指派一个唯一的ID（从1开始），一旦消息被投递到所有匹配的队列之后；

rabbitMQ就会发送一个ACK给生产者（包含消息的唯一ID），这就使得生产者知道消息已经正确到达目的队列了；

如果rabbitMQ没能处理该消息，则会发送一个Nack消息给你，可以进行重试操作。

**消息列表丢失消息:**

可以消息持久化, 即使rabbitMQ挂了，重启后也能恢复数据

**消费者丢失消息****:**

消费者丢数据一般是因为采用了自动确认消息模式，消费者在收到消息之后，处理消息之前，会自动回复RabbitMQ已收到消息；如果这时处理消息失败，就会丢失该消息；改为手动确认消息即可！

## **4、消息重复消费问题** ##

**为什么会重复消费：**

正常情况下，消费者在消费消息的时候，消费完毕后，会发送一个确认消息给消息队列，消息队列就知道该消息被消费了，就会将该消息从消息队列中删除；

但是因为网络传输等等故障，确认信息没有传送到消息队列，导致消息队列不知道已经消费过该消息了，再次将消息发送。

**解决方案：**保证消息的唯一性，就算是多次传输，不要让消息的多次消费带来影响，保证消息消费的幂等性；

## **5、幂等性操作** ##

幂等性就是一个数据或者一个请求，给你重复来了多次，你得确保对应的数据是不会改变的，不能出错。

要保证消息的幂等性，这个要结合业务的类型来进行处理。

1）、可在内存中维护一个map集合，只要从消息队列里面消费一个消息，先查询这个消息在不在map里面，如果在表示已消费过，直接丢弃；如果不在，则在消费后将其加入map当中。  
2）、如果要写入数据库，可以拿唯一键先去数据库查询一下，如果不存在在写，如果存在直接更新或者丢弃消息。

3）、消息执行完会更改某个数据状态，判断数据状态是否更新，如果更新，则不进行重复消费。  
4）、如果是写redis那没有问题，每次都是set，天然的幂等性。

[12a5163627fb4f99a78bdf4ba4a10894.png]: https://img-blog.csdnimg.cn/12a5163627fb4f99a78bdf4ba4a10894.png
[244d8ca3f524443d97de0160fe9d7510.png]: https://img-blog.csdnimg.cn/244d8ca3f524443d97de0160fe9d7510.png
[d50dd3c01d8b48d1bc8ea09b6bc0ee9d.png]: https://img-blog.csdnimg.cn/d50dd3c01d8b48d1bc8ea09b6bc0ee9d.png
[9a786426641442c4a4bc7bf72d1da011.png]: https://img-blog.csdnimg.cn/9a786426641442c4a4bc7bf72d1da011.png
[f40e7f44cd2b4f0fade1c3726bd9b6f2.png]: https://img-blog.csdnimg.cn/f40e7f44cd2b4f0fade1c3726bd9b6f2.png
[96f57d5d38b745bfaddd777526b0ac6d.png]: https://img-blog.csdnimg.cn/96f57d5d38b745bfaddd777526b0ac6d.png