RabbitMQ[复盘]

Flaw2024-08-092025-05-07

1.1概述

在项目中，可将一些无需即时返回且耗时的操作提取出来，进行异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了**系统的吞吐量。

1.1.1 QPS概念

QPS即每秒查询率，是对一个特定的查询服务器在规定时间内所处理流量多少的衡量标准。
每秒查询率：
因特网上，经常用每秒查询率来衡量域名系统服务器的机器的性能，即为QPS。
或者理解：每秒的响应请求数，也即是最大吞吐能力。

计算关系：
QPS = 并发量 / 平均响应时间
并发量 = QPS * 平均响应时间
原理：每天80%的访问集中在20%的时间里，这20%时间叫做峰值时间
公式：( 总PV数 * 80% ) / ( 每天秒数 * 20% ) = 峰值时间每秒请求数(QPS) 。
机器：峰值时间每秒QPS / 单台机器的QPS = 需要的机器。

1.1.2PV UV PR

① 什么是pv
**PV(page view)，即页面浏览量，或点击量；通常是衡量一个网络新闻频道或网站甚至一条网络新闻的主要指标。
对pv的解释是，一个访问者在24小时(0点到24点)内到底看了你网站几个页面。这里需要强调：同一个人浏览你网站同一个页面，不重复计算pv量，点100次也算1次。说白了，pv就是一个访问者打开了你的几个页面。
PV之于网站，就像收视率之于电视，从某种程度上已成为投资者衡量商业网站表现的最重要尺度。
pv的计算：当一个访问者访问的时候，记录他所访问的页面和对应的IP，然后确定这个IP今天访问了这个页面没有。如果你的网站到了23点，单纯IP有60万条的话，每个访问者平均访问了3个页面，那么pv表的记录就要有180万条。
② 什么是uv
** uv(unique visitor)，指访问某个站点或点击某条新闻的不同IP地址的人数。
在同一天内，uv只记录第一次进入网站的具有独立IP的访问者，在同一天内再次访问该网站则不计数。独立IP访问者提供了一定时间内不同观众数量的统计指标，而没有反应出网站的全面活动。
③ 什么是pr值
PR值，即PageRank，网页的级别技术，用来标识网页的等级/重要性。级别从1到10级，10级为满分。PR值越高说明该网页越受欢迎（越重要）。
例如：一个PR值为1的网站表明这个网站不太具有流行度，而PR值为7到10则表明这个网站非常受欢迎（或者说极其重要）。

1.2.消息队列应用场景

1.2.1应用解耦

传统模式的缺点：

系统间耦合性太强，如上图所示，系统A在代码中直接调用系统B和系统C的代码，如果将来D系统接入，系统A还需要修改代码，过于麻烦！

中间件模式：

将消息写入消息队列，需要消息的系统自己从消息队列中订阅，从而系统A不需要做任何修改。

1.2.2.异步处理

用户注册后，需要发注册邮件和注册短信，传统的做法有两种

  - 串行的方式
  - 并行的方式

(1) 串行方式：
将注册信息写入数据库后，发送注册邮件，再发送注册短信，以上三个任务全部完成后才返回给客户端。这有一个问题是，邮件，短信并不是必须的，它只是一个通知，而这种做法让客户端等待没有必要等待的东西。

(2) 并行方式：
将注册信息写入数据库后，发送邮件的同时，发送短信，以上三个任务完成后，返回给客户端，并行的方式能提高处理的时间。

(3)消息队列：

1.2.3.流量削峰

流量削峰一般在秒杀活动中应用广泛
** 场景：** 秒杀活动，一般会因为流量过大，导致应用挂掉，为了解决这个问题，一般在应用前端加入消息队列。
如订单系统，在下单的时候就会往数据库写数据。但是数据库只能支撑每秒1000左右的并发写入，并发量再高就容易宕机。低峰期的时候并发也就100多个，但是在高峰期时候，并发量会突然激增到5000以上，这个时候数据库肯定卡死了。

传统模式的缺点：

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

中间件模式：

消息被MQ保存起来了，然后系统就可以按照自己的消费能力来消费，比如每秒1000个数据，这样慢慢写入数据库，这样就不会卡死数据库了。
系统A慢慢的按照数据库能处理的并发量，从消息队列中慢慢拉取消息。在生产中，这个短暂的高峰期积压是允许的。

2.消息队列相关介绍

MQ是消息通信的模型；实现MQ的大致有两种主流方式：AMQP、JMS。

2.2.1. AMQP

AMQP是一种高级消息队列协议（Advanced Message Queuing Protocol），更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。

2.2.2. JMS

JMS即Java消息服务（JavaMessage Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

2.2.3. AMQP 与 JMS 区别

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模式；而AMQP的消息模式更加丰富

2.2.4 RabbitMQ简介

AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。
消费积压

2.2.5 RabbitMQ 中的相关概念

Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker
Virtual host:出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
Connection：publisher／consumer 和 broker 之间的 TCP 连接
Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销
Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：direct (point-to-point)，topic (publish-subscribe) and fanout (multicast)
Queue：存储消息的容器，消息最终被送到这里，等待 consumer 取走
Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

3.RabbitMQ安装及配置

3.1下载包

1.下载Erlang的rpm包
RabbitMQ是Erlang语言编写，所以Erang环境必须要有，注：Erlang环境一定要与RabbitMQ版本匹配：
https://www.rabbitmq.com/which-erlang.html
2.下载socat的rpm包
rabbitmq安装依赖于socat，所以需要下载socat。
socat下载地址：http://repo.iotti.biz/CentOS/7/x86_64/socat-1.7.3.2-5.el7.lux.x86_64.rpm
3.下载RabbitMQ的rpm包
RabbitMQ下载地址：https://www.rabbitmq.com/download.html（根据自身需求及匹配关系，下载对应rpm包）rabbitmq-server-3.8.1-1.el7.noarch.rpm
也可从github等网站下载。https://hub.fastgit.org/rabbitmq/rabbitmq-server/releases/

3.2安装

1.安装Erlang、Socat、RabbitMQ

**①rpm -ivh erlang-21.3.8.9-1.el7.x86_64.rpm **
②rpm -ivh socat-1.7.3.2-1.el6.lux.x86_64.rpm
在安装rabbitmq之前需要先安装socat，否则，报错。可以采用yum安装方式：yum install socat，我们这里采用rpm安装方式
**③rpm -ivh rabbitmq-server-3.8.1-1.el7.noarch.rpm **
/usr/lib/rabbitmq/bin/

2.启用管理插件

rabbitmq-plugins enable rabbitmq_management

3.启动RabbitMQ

systemctl start rabbitmq-server.service
systemctl status rabbitmq-server.service
systemctl restart rabbitmq-server.service
systemctl stop rabbitmq-server.service

4.查看进程

ps -ef | grep rabbitmq

5.测试

  - 关闭防火墙：systemctl stop firewalld.service
  -  在web浏览器中输入地址：http://虚拟机ip:15672/
  - 输入默认账号密码：guest ： guest，guest用户默认不允许远程连接。

6.增加自定义账号

  1. 添加管理员账号密码：rabbitmqctl add_user admin admin
  2. 分配账号角色：rabbitmqctl set_user_tags admin administrator
  3. 设置用户权限 ：set_permissions [-p <vhostpath>] <user> <conf> <write> <read> 

rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*" 

  4. 用户user_admin具有/vhost1这个virtual host中所有资源的配置、写、读权限
  5. 修改密码：rabbitmqctl change_password admin 123456
  6. 查看用户列表：rabbitmqctl list_users

管理界面标签页介绍
overview：概览
connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况
channels：通道，建立连接后，会形成通道，消息的投递获取依赖通道。
Exchanges：交换机，用来实现消息的路由
Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列。
** 端口：**
5672：rabbitMq的编程语言客户端连接端口
15672：rabbitMq管理界面端口
25672：rabbitMq集群的端口

7.卸载

rpm -qa | grep rabbitmq
rpm -e rabbitmq-server

3.3管理界面

虚拟主机：类似于mysql中的database。他们都是以“/”开头

4.AMQP

4.1. 相关概念介绍

AMQP 一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。
RabbitMQ是AMQP协议的Erlang的实现。

概念	说明
连接Connection	一个网络连接，比如TCP/IP套接字连接。
信道Channel	多路复用连接中的一条独立的双向数据流通道。为会话提供物理传输介质。
客户端Client	AMQP连接或者会话的发起者。AMQP是非对称的，客户端生产和消费消息，服务器存储和路由这些消息。
服务节点Broker	消息中间件的服务节点；一般情况下可以将一个RabbitMQ Broker看作一台RabbitMQ 服务器。
端点	AMQP对话的任意一方。一个AMQP连接包括两个端点（一个是客户端，一个是服务器）。
消费者Consumer	一个从消息队列里请求消息的客户端程序。
生产者Producer	一个向交换机发布消息的客户端应用程序。

4.2. RabbitMQ运转流程

在入门案例中：
· 生产者发送消息

生产者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker；
声明队列并设置属性；如是否排它，是否持久化，是否自动删除；
将路由键（空字符串）与队列绑定起来；
发送消息至RabbitMQ Broker；
关闭信道；
关闭连接；
· 消费者接收消息
消费者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker
向Broker 请求消费相应队列中的消息，设置相应的回调函数；
等待Broker投递响应队列中的消息，消费者接收消息；
确认（ack，自动确认）接收到的消息；
RabbitMQ从队列中删除相应已经被确认的消息；
关闭信道；
关闭连接；

5.RabbitMQ入门案列

5.1入门案列

简单生产者消费者

<dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

package com.atguigu.rabbitmq.simple;
/*
 *1.生产者
 */

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.10.101");
        connectionFactory.setPort(5672);

        connectionFactory.setVirtualHost("/");

        connectionFactory.setUsername("admin");
        //连接密码；默认为guest
        connectionFactory.setPassword("admin");

        Connection connection = connectionFactory.newConnection();

        Channel channel = connection.createChannel();
        /**
         * queue      参数1：队列名称
         * durable    参数2：是否定义持久化队列,当mq重启之后,还在
         * exclusive  参数3：是否独占本次连接
         *            ① 是否独占,只能有一个消费者监听这个队列
         *            ② 当connection关闭时,是否删除队列
         * autoDelete 参数4：是否在不使用的时候自动删除队列,当没有consumer时,自动删除
         * arguments  参数5：队列其它参数
         */

        // 声明（创建）队列
        channel.queueDeclare("simple_queue",true,false,false,null);

        // 要发送的信息
        String message = "你好；小兔子！";
        /**
         * 参数1：交换机名称,如果没有指定则使用默认Default Exchage
         * 参数2：路由key,简单模式可以传递队列名称
         * 参数3：配置信息
         * 参数4：消息内容
         */

        channel.basicPublish("","simple_queue",null,message.getBytes());
        System.out.println("已发送消息：" + message);

        // 关闭资源
        channel.close();
        connection.close();
    }
}

package com.atguigu.rabbitmq.simple;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory  connectionFactory= new ConnectionFactory();

        connectionFactory.setHost("192.168.10.101");
        connectionFactory.setPort(5672);

        connectionFactory.setVirtualHost("/");

        connectionFactory.setUsername("admin");
        //连接密码；默认为guest
        connectionFactory.setPassword("admin");

        Connection connection = connectionFactory.newConnection();

        Channel channel = connection.createChannel();

        //5. 创建队列Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        参数：
            1. queue：队列名称
            2. durable：是否持久化,当mq重启之后,还在
            3. exclusive：
                * 是否独占。只能有一个消费者监听这队列
                * 当Connection关闭时,是否删除队列
            4. autoDelete：是否自动删除。当没有Consumer时,自动删除掉
            5. arguments：参数。
         */
        //如果没有一个名字叫simple_queue的队列,则会创建该队列,如果有则不会创建
        channel.queueDeclare("simple_queue",true,false,false,null);

        // 接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);
                System.out.println("body："+new String(body));
            }
        };
        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
        1. queue：队列名称
        2. autoAck：是否自动确认 ,类似咱们发短信,发送成功会收到一个确认消息
        3. callback：回调对象
        */
        // 消费者类似一个监听程序,主要是用来监听消息
        channel.basicConsume("simple_queue",true,consumer);

    }
}

6.RabbitMQ工作模式

6.1 简单模式

6.1.1概述

多个消费端共同消费同一个队列中的消息。
消费者之间是竞争的关系
应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度

6.1.2 代码实现

** 一个生产者，一个消费者**

public class ConnectionRabbitMQUtil {
    public static Connection getConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.10.101");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");

        //获取链接
        Connection connection = factory.newConnection();
        return  connection;
    }

    public static  void release(Connection connection,Channel channel){
        try {
            channel.close();
            connection.close();
        }  catch (TimeoutException | IOException e) {
            throw new RuntimeException(e);
        }
    }
}

package com.atguigu.rabbitmq.simple;
import com.atguigu.rabbitmq.utils.ConnectionRabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer_01 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        //1.创建信道
        Channel channel = connection.createChannel();

        //2.声明队列
        channel.queueDeclare("worker_queue",true,false,false,null);

        //3.发消息
        for (int i = 0;i<100000000; i++) {
            String body = i+"hello rabbitmq~~~";
            channel.basicPublish("","worker_queue",null,body.getBytes());
        }
        System.out.println("发送完毕.......");
        //4.释放链接
        ConnectionRabbitMQUtil.release(connection,channel);

    }
}

package com.atguigu.rabbitmq.simple;
import com.atguigu.rabbitmq.utils.ConnectionRabbitMQUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer_01 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();

        //1.声明队列
        channel.queueDeclare("worker_queue",true,false,false,null);

        //2.实现怎样消费
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
            }
        };

        //3.消费
        channel.basicConsume("worker_queue",true,consumer);


    }
}

6.2 发布订阅模式

6.2.1模式说明

发布订阅模式：
1、每个消费者监听自己的队列。
2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息

6.2.2 代码实现

/*
 * 发布订阅模式
 */
public class Provider {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();

        //1.创建交换机
        String exchangeName = "test_fanout";
        //发布订阅模式
       /*
        exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
        参数：
        1. exchange：交换机名称
        2. type：交换机类型
            DIRECT("direct"),：定向
            FANOUT("fanout"),：扇形（广播）,发送消息到每一个与之绑定队列。
            TOPIC("topic"),通配符的方式
            HEADERS("headers");参数匹配
        3. durable：是否持久化
        4. autoDelete：自动删除
        5. internal：内部使用。 一般false
        6. arguments：参数
        */
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);

        //2. 创建队列
        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);


        //3.绑定队列与交换机
        channel.queueBind(queue1Name,exchangeName,"");
        channel.queueBind(queue2Name,exchangeName,"");


        //4.发送消息
        String body = "日志信息：张三调用了findAll方法...日志级别：info...";
        channel.basicPublish(exchangeName,"",null,body.getBytes());

        //5. 释放资源
        channel.close();
        connection.close();
    }
}

public class Customer_1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        String queue1Name = "test_fanout_queue1";
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body："+new String(body));
                System.out.println("将日志信息打印到控制台.....");
            }
        };
        channel.basicConsume(queue1Name,true,consumer);
    }
}

public class Customer_2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        String queue1Name = "test_fanout_queue2";
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body："+new String(body));
                System.out.println("将日志信息打印到控制台.....");
            }
        };
        channel.basicConsume(queue1Name,true,consumer);
    }
}

6.3Routing路由模式

6.3.1模式说明

路由模式特点：

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。
Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

· P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
· X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列
· C1：消费者，其所在队列指定了需要routing key 为 error 的消息
· C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

6.3.2代码实现

package com.atguigu.rabbitmq.direct;
import com.atguigu.rabbitmq.utils.ConnectionRabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();

        //1.创建交换机
                     
        //发布订阅模式
       /*
        exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
        参数：
        1. exchange：交换机名称
        2. type：交换机类型
            DIRECT("direct"),：定向
            FANOUT("fanout"),：扇形（广播）,发送消息到每一个与之绑定队列。
            TOPIC("topic"),通配符的方式
            HEADERS("headers");参数匹配
        3. durable：是否持久化
        4. autoDelete：自动删除
        5. internal：内部使用。 一般false
        6. arguments：参数
        */
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);

        //2. 创建队列
        String queue1Name = "test_direct_queue1";
        String queue2Name = "test_direct_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);


        //3.队列绑定交换机
        channel.queueBind(queue1Name,exchangeName,"error"); //指定路由key

        // 队列2绑定info error warning
        channel.queueBind(queue2Name,exchangeName,"info");
        channel.queueBind(queue2Name,exchangeName,"error");
        channel.queueBind(queue2Name,exchangeName,"warning");

        String message = "日志信息：张三调用了delete方法.错误了,日志级别warning";

        channel.basicPublish(exchangeName,"info",null,message.getBytes());
        System.out.println(message);

        channel.close();
        connection.close();
    }
}

package com.atguigu.rabbitmq.direct;

import com.atguigu.rabbitmq.utils.ConnectionRabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer_01 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        String queue1Name = "test_direct_queue1";
        String queue2Name = "test_direct_queue2";

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body："+new String(body));
                System.out.println("将日志信息打印到控制台.....");
            }
        };
        channel.basicConsume(queue1Name,true,consumer);

    }
}

public class Consumer_02 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        String queue1Name = "test_direct_queue1";
        String queue2Name = "test_direct_queue2";

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body："+new String(body));
                System.out.println("将日志信息打印到控制台.....");
            }
        };
        channel.basicConsume(queue2Name,true,consumer);

    }
}

6.4 Topics通配符模式

6.4.1模式说明

Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！
Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
通配符规则：
#：匹配零个或多个词
*：匹配不多不少恰好1个词
举例：
item.#：能够匹配item.insert.abc 或者 item.insert
item.*：只能匹配item.insert

6.4.2 代码实现

package com.atguigu.rabbitmq.topic;
import com.atguigu.rabbitmq.utils.ConnectionRabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Provider {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();

        //1.创建交换机
        String exchangeName = "test_topic";
        //发布订阅模式
       /*
        exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
        参数：
        1. exchange：交换机名称
        2. type：交换机类型
            DIRECT("direct"),：定向
            FANOUT("fanout"),：扇形（广播）,发送消息到每一个与之绑定队列。
            TOPIC("topic"),通配符的方式
            HEADERS("headers");参数匹配
        3. durable：是否持久化
        4. autoDelete：自动删除
        5. internal：内部使用。 一般false
        6. arguments：参数
        */
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);

        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);
        // 绑定队列和交换机
        /**
         *  参数：
         1. queue：队列名称
         2. exchange：交换机名称
         3. routingKey：路由键,绑定规则
         如果交换机的类型为fanout ,routingKey设置为""
         */
        // routing key  系统的名称.日志的级别。
        //需求： 所有error级别的日志存入数据库,所有order系统的日志存入数据库
        channel.queueBind(queue1Name,exchangeName,"#.error");
        channel.queueBind(queue2Name,exchangeName,"order.*");
        String body1 = "队列1.....";
        //发送消息goods.info,goods.error
        channel.basicPublish(exchangeName,"order.error",null,body1.getBytes());
        channel.close();
        connection.close();
        }
}

package com.atguigu.rabbitmq.topic;
import com.atguigu.rabbitmq.utils.ConnectionRabbitMQUtil;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer1 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        String queue1Name = "test_topic_queue1";
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body："+new String(body));
            }
        };
        channel.basicConsume(queue1Name,true,consumer);
    }
}

public class Consumer2 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionRabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        String queue1Name = "test_topic_queue2";
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body："+new String(body));
            }
        };
        channel.basicConsume(queue1Name,true,consumer);
    }
}

6.5工作模式总结

RabbitMQ工作模式：
1、简单模式 HelloWorld
一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

2、工作队列模式 Work Queue
一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）

3、发布订阅模式 Publish/subscribe
需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列

4、路由模式 Routing
需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

5、通配符模式 Topic
需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

7.Spring整合RabbitMQ

7.1 生产者端

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>_07RabbitMQ</artifactId>
        <groupId>com.atguigu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>spring-rabbitmq-proverder</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>


    <dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>2.1.8.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>

        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>


</project>

rabbitmq.host=192.168.10.101
rabbitmq.port=5672
rabbitmq.username=admin
rabbitmq.password=admin
rabbitmq.virtual-host=/

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    <!--加载配置文件-->
    <context:property-placeholder location="classpath:rabbit.properties"/>

    <!-- 定义rabbitmq connectionFactory -->
    <rabbit:connection-factory id="connectionFactory"
                               host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"/>
    <!-- 定义RabbitAdmin对象，用于管理交换机、队列-->
        <rabbit:admin connection-factory="connectionFactory"/>
    
    <!--
        1.简单模式
        定义持久化队列,不存在则自动创建；不绑定到交换机则绑定到默认交换机
        默认交换机类型为direct,名字为："",路由键为队列的名称
    -->
        <rabbit:queue id="spring_queue" name="spring-simple-queue" auto-declare="true"/>

        <!-- 2.广播模式   -->
        <rabbit:queue id="spring-fanout-queue-1" name="spring-fanout-queue-11" auto-declare="true"/>
        <rabbit:queue id="spring-fanout-queue-2" name="spring-fanout-queue-22" auto-declare="true"/>
        
        <!-- 绑定队列与交换机   -->
        <rabbit:fanout-exchange name="spring-fanout-exchange" id="spring-fanout-exchange" auto-declare="true">
              <rabbit:bindings>
                  <rabbit:binding queue="spring-fanout-queue-1"/>
                  <rabbit:binding queue="spring-fanout-queue-2"/>
              </rabbit:bindings>
        </rabbit:fanout-exchange>


    <!-- 3.主题（通配符）模式   -->
        <!--定义主题交换机中的持久化队列,不存在则自动创建-->
        <rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star" auto-declare="true"/>
        <!--定义主题交换机中的持久化队列,不存在则自动创建-->
        <rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/>
        <!--定义主题交换机中的持久化队列,不存在则自动创建-->
        <rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/>
      
       <rabbit:topic-exchange name="spring_topic_exchange" id="spring_topic_exchange" auto-declare="true">
            <rabbit:bindings>
                <rabbit:binding pattern="atguigu.*" queue="spring_topic_queue_star"/>
                <rabbit:binding pattern="atguigu.#" queue="spring_topic_queue_well"/>
                <rabbit:binding pattern="guigu.#" queue="spring_topic_queue_well2"/>
            </rabbit:bindings>
        </rabbit:topic-exchange>

    <!--4.定义rabbitTemplate对象操作可以在代码中方便发送消息-->
        <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
</beans>

package com.atguigu;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq.xml")
public class ProducerTest {


    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * 1.简单队列消息
     * 默认交换机类型为 direct
     * 交换机的名称为空,路由键为队列的名称
     */
    @Test
    public void queueTest(){
        //路由键与队列同名
        rabbitTemplate.convertAndSend("spring_simple_queue", "只发队列spring_sinple_queue的消息。");
        System.out.println("消息发送完毕........");
    }

    /**
     * 广播模式
     */
    @Test
    public void fanoutTest(){
        /**
         * 参数1：交换机名称
         * 参数2：路由键名（广播设置为空）
         * 参数3：发送的消息内容
         */
        while (true)
          rabbitTemplate.convertAndSend("spring-fanout-exchange", "", "发送到spring_fanout_exchange交换机的广播消息");
    }


    /**
     * 主题模式
     */
    @Test
    public void TopicTest(){
        /**
         * 参数1：交换机名称
         * 参数2：路由键名（广播设置为空）
         * 参数3：发送的消息内容
         */
        rabbitTemplate.convertAndSend("spring_topic_exchange", "", "发送到spring_topic_exchange交换机的主题消息");
    }

    /**
     * 通配符
     * 交换机类型为 topic
     * 匹配路由键的通配符,*表示一个单词,#表示多个单词
     * 绑定到该交换机的匹配队列能够收到对应消息
     */
    @Test
    public void topicTest(){
        /**
         * 参数1：交换机名称
         * 参数2：路由键名
         * 参数3：发送的消息内容
         */
        rabbitTemplate.convertAndSend("spring_topic_exchange", "atguigu.bj", "发送到spring_topic_exchange交换机atguigu.bj的消息");
        rabbitTemplate.convertAndSend("spring_topic_exchange", "guigu.cn", "发送到spring_topic_exchange交换机guigu.cn的消息");
    }

}

7.2.消费者端

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    <!--加载配置文件-->
    <context:property-placeholder location="classpath:rabbit.properties"/>

    <!-- 定义rabbitmq connectionFactory -->
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"/>
    <!-- 定义RabbitAdmin对象，用于管理交换机、队列-->
    <rabbit:admin connection-factory="connectionFactory"/>

    <!-- 1.配置简单模式队列监听器队列   -->
    <bean id="springQueueListener" class="com.atguig.Listener.SpringQueueListener" />
    <!-- 2.广播队列-->
    <bean id="springFanoutListener1" class="com.atguig.Listener.SpringFanoutQueue1Listener" />
    <bean id="springFanoutListener2" class="com.atguig.Listener.SpringFanoutQueue2Listener" />
    <!--3.主题队列-->
    <bean id="springTopic1Listener" class="com.atguig.Listener.SpringTopic1Listener"/>
    <bean id="springTopic2Listener" class="com.atguig.Listener.SpringTopic2Listener"/>

   <!-- 配置监听器   -->
   <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true">
        <!-- 简单模式配置       -->
        <rabbit:listener ref="springQueueListener" queue-names="spring_simple_queue"/> //消费队列信息
        <!-- 广播配置      -->
        <rabbit:listener ref="springFanoutListener1" queue-names="spring-fanout-queue-11"/>
        <rabbit:listener ref="springFanoutListener2" queue-names="spring-fanout-queue-22"/>
        <!-- 主题（通配符）队列       -->
        <rabbit:listener ref="springTopic1Listener" queue-names="spring_topic_queue_star"/>
        <rabbit:listener ref="springTopic2Listener" queue-names="spring_topic_queue_well2"/>
   </rabbit:listener-container>

</beans>

/**
 * 简单队列监听
 */
public class SpringQueueListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        try {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.printf("接收路由名称为：%s,路由键为：%s,队列名为：%s的消息：%s \n",
                    message.getMessageProperties().getReceivedExchange(),
                    message.getMessageProperties().getReceivedRoutingKey(),
                    message.getMessageProperties().getConsumerQueue(),
                    msg);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq.xml")
public class CustomerTests {


    @Test
    public void test1(){
        boolean flag = true;
        while (flag){

        }
    }

}

8.RabbitMQ高级特性

8.1消息的可靠性传递（发送端）

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。
· confirm 确认模式
· return 退回模式
rabbitmq 整个消息投递的路径为：
producer—>rabbitmq broker—>exchange—>queue—>consumer

     1. 消息从 producer 到 exchange 则会返回一个 confirmCallback 。
     2. 消息从 exchange–>queue 投递失败则会返回一个 returnCallback 。

8.1.1 Confirm确认模式

** 装配时要配置confirm为true** publisher-confirms="true" publisher-returns="true"

<dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>2.1.8.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>

        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>
    </dependencies>

rabbitmq.host=192.168.10.101
rabbitmq.port=5672
rabbitmq.username=admin
rabbitmq.password=admin
rabbitmq.virtual-host=/

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    <!--加载配置文件-->
    <context:property-placeholder location="classpath:rabbit.properties"/>

    <!-- 定义rabbitmq connectionFactory -->
    <rabbit:connection-factory id="connectionFactory"  publisher-confirms="true" publisher-returns="true"  //设置为true
                               host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"/>
    <!-- 定义RabbitAdmin对象，用于管理交换机、队列-->
    <rabbit:admin connection-factory="connectionFactory"/>

    <!--绑定队列-->
    <rabbit:queue id="test_queue_confirm" name="test_queue_confirm" auto-declare="true"/>
    <rabbit:direct-exchange name="test_exchange_confirm" id="test_exchange_confirm" auto-declare="true">
             <rabbit:bindings>
                 <rabbit:binding key="confirm" queue="test_queue_confirm"/>
             </rabbit:bindings>
    </rabbit:direct-exchange>

    <!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
</beans>

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void queueTest(){
       rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
           @Override
           public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack){
                    System.out.println("接收消息成功："+cause);
                }
                else{
                    //接收失败
                    System.out.println("接收失败消息" + cause); 
                    //做一些处理,让消息再次发送。
                }
           }
       });


//       rabbitTemplate.convertAndSend("test_exchange_confirm","confirmss","message confirm".getBytes());//成功
        rabbitTemplate.convertAndSend("test_exchange_confirm000", "confirm", "message confirm....");//失败
        //接收失败消息channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'test_exchange_confirm000' in vhost '/', class-id=60, method-id=40)
    }

8.1.2 Return退回模式

/**
   * 回退模式： 当消息发送给Exchange后,Exchange路由到Queue失败时 才会执行 ReturnCallBack
   * 步骤：
   * 1. 开启回退模式：publisher-returns="true"
   * 2. 设置ReturnCallBack
   * 3. 设置Exchange处理消息的模式：
   *      1). 如果消息没有路由到Queue,则丢弃消息（默认）
   *      2). 如果消息没有路由到Queue,返回给消息发送方ReturnCallBack
   *            rabbitTemplate.setMandatory(true);
   */
  @Test
  public void testReturn() {
      //设置交换机处理失败消息的模式
      rabbitTemplate.setMandatory(true);
      //2.设置ReturnCallBack
      rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
          /**
           * @param message   消息对象
           * @param replyCode 错误码
           * @param replyText 错误信息
           * @param exchange  交换机
           * @param routingKey 路由键
           */
          @Override
          public void returnedMessage(Message message, int replyCode,String replyText,String exchange,String routingKey) {
              System.out.println("return 执行了....");

              System.out.println(message);
              System.out.println(replyCode);
              System.out.println(replyText);
              System.out.println(exchange);
              System.out.println(routingKey);

              //处理
          }
      });
      //3. 发送消息
      rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm4", "message confirm....");//失败
      /*
      return 执行了....
      (Body:'message confirm....' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
      312
      NO_ROUTE
      test_exchange_confirm
      confirm4
       */

  }

8.1.3消息的可靠投递小结

设置 ConnectionFactory的publisher-confirms=”true” 开启确认模式。
使用 rabbitTemplate.setConfirmCallback 设置回调函数。当消息发送到 exchange 后回调 confirm 方法。在方法中判断 ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。
设置 ConnectionFactory 的 publisher-returns=”true” 开启退回模式。
使用 rabbitTemplate.setReturnCallback 设置退回函数，当消息从exchange 路由到 queue 失败后，如果设置了 rabbitTemplate.setMandatory(true) 参数，则会将消息退回给 producer并执行回调函数returnedMessage

8.1.4Consumer Ack（接收端）

ack指Acknowledge，确认。表示消费端收到消息后的确认方式。

有二种确认方式：
· 自动确认：acknowledge=“none” 默认
· 手动确认：acknowledge=“manual”

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。
如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

<dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>2.1.8.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>

        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>
    </dependencies>

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

    <context:component-scan base-package="com.atguig" />
    
    <!--加载配置文件-->
    <context:property-placeholder location="classpath:rabbit.properties"/>

    <!-- 定义rabbitmq connectionFactory -->
    <rabbit:connection-factory id="connectionFactory" 
                               host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"/>
    <!-- 定义RabbitAdmin对象，用于管理交换机、队列-->
    <rabbit:admin connection-factory="connectionFactory"/>

    <!--绑定队列-监听器-->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">   <!--设置手动确认 -->
          <rabbit:listener ref="ackListener" queue-names="test_queue_confirm"/>    <!--设置监听的队列 -->
    </rabbit:listener-container>  
                                 
    <!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
</beans>

package com.atguig.Listener;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

/**
 * Consumer ACK机制：
 *  1. 设置手动签收。acknowledge="manual"
 *  2. 让监听器类实现ChannelAwareMessageListener接口
 *  3. 如果消息成功处理,则调用channel的 basicAck()签收
 *  4. 如果消息处理失败,则调用channel的basicNack()拒绝签收,broker重新发送给consumer
 */
@Component
public class AckListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        Thread.sleep(1000);
        // 获取消息传递标记
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            System.out.println("deliveryTag:"+deliveryTag);
            // ① 接收消息
            System.out.println(new String(message.getBody()));
            // ② 处理业务逻辑
            System.out.println("处理业务逻辑");
//            int i = 3/0;//出现错误
            // ③ 手动签收
            /**
             * 第一个参数：表示收到的标签
             * 第二个参数：如果为true表示可以签收所有的消息
             */
            channel.basicAck(deliveryTag,true);  //手动提交
        } catch (Exception e) {
            e.printStackTrace();
            // ④ 拒绝签收
             /*
            第三个参数：requeue：重回队列。
            设置为true,则消息重新回到queue,broker会重新发送该消息给消费端
             */
            channel.basicNack(deliveryTag,true,true);
        }

    }
}

8.2消费限流

prefetch = 1,表示消费端每次从mq拉去一条消息来消费,直到手动确认消费完毕后,才会继续拉取下一条消息。
在 rabbit:listener-container中配置 prefetch 属性设置消费端一次拉取多少条消息
消费端的必须确认才会继续处理其他消息。

8.3TTL

TTL 全称 Time To Live（存活时间/过期时间）。
当消息到达存活时间后，还没有被消费，会被自动清除。
** RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。**

8.3.1 配置队列过期时间

<!--绑定队列-->
<rabbit:queue id="test_queue_confirm" name="test_queue_confirm" auto-declare="true">
     <rabbit:queue-arguments>
          <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
     </rabbit:queue-arguments>
</rabbit:queue>

8.3.2 消息过期时间

/**
    * TTL：过期时间
    *  1. 队列统一过期
    *  2. 消息单独过期
    * 如果设置了消息的过期时间,也设置了队列的过期时间,它以时间短的为准。
    */
   @Test
   public void testMessageTtl() {
       // 消息后处理对象,设置一些消息的参数信息
       MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {

           @Override
           public Message postProcessMessage(Message message) throws AmqpException {
               //1.设置message的信息
               // 第二个方法：消息的过期时间 ,5秒之后过期
               message.getMessageProperties().setExpiration("5000");
               //2.返回该消息
               return message;
           }
       };

       //消息单独过期
       rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl....",messagePostProcessor);
   }

8.4死信队列

死信队列，英文缩写：DLX 。DeadLetter Exchange（死信交换机），当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX。

8.4.1什么是死信队列

死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer将消息投递到broker或者直接到queue里了，consumer从queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信，自然就有了死信队列；

消息成为死信的三种情况：
<1> 队列消息数量到达限制；比如队列最大只能存储10条消息，而发了11条消息，根据先进先出，最先发的消息会进入死信队列。
<2>消费者拒接消费消息，basicNack/basicReject，并且不把消息重新放入原目标队列，requeue=false；
<3>原队列存在消息过期设置，消息到达超时时间未被消费；

死信的处理方式:
死信的产生既然不可避免，那么就需要从实际的业务角度和场景出发，对这些死信进行后续的处理，常见的处理方式:
① 丢弃，如果不是很重要，可以选择丢弃
② 记录死信入库，然后做后续的业务分析或处理
③ 通过死信队列，由负责监听死信的应用程序进行处理

队列绑定死信交换机：
给队列设置参数：x-dead-letter-exchange 和x-dead-letter-routing-key

8.4.2 过期时间处理代码

 <!-- 1.正常队列   -->
 <rabbit:queue id="test_queue_dlx" name="test_queue_dlx">
     <!--  绑定死信队列        -->
     <rabbit:queue-arguments>
         <!--3.1 x-dead-letter-exchange：死信交换机名称-->
         <entry key="x-dead-letter-exchange" value="exchange_dlx"/>

         <!--3.2 x-dead-letter-routing-key：发送给死信交换机的routingkey-->
         <entry key="x-dead-letter-routing-key" value="dlx.hehe"/>

         <!--4.1 设置队列的过期时间10s ttl-->
         <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
         
      </rabbit:queue-arguments>
 </rabbit:queue>
 
 <!-- 2.正常交换机   -->
<rabbit:topic-exchange name="test_exchange_dlx" id="test_exchange_dlx">
      <rabbit:bindings>
          <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"/>
      </rabbit:bindings>
</rabbit:topic-exchange>


 <!--3.死信队列    -->
<rabbit:queue id="queue_dlx" name="queue_dlx"/>

 <!--4.死信交换机    -->
 <rabbit:topic-exchange name="exchange_dlx" id="exchange_dlx">
      <rabbit:bindings>
          <rabbit:binding pattern="dlx.#" queue="queue_dlx"/>
      </rabbit:bindings>
 </rabbit:topic-exchange>

8.4.3 长度限制处理

 <!-- ####################################死信队列测试###################################################################-->
 <!--
     死信队列：
         1. 声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)
         2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx)
         3. 正常队列绑定死信交换机
             设置两个参数：
                 * x-dead-letter-exchange：死信交换机名称
                 * x-dead-letter-routing-key：发送给死信交换机的routingkey
 -->
 <!-- 1.正常队列   -->
 <rabbit:queue id="test_queue_dlx" name="test_queue_dlx">
     <!--  绑定死信队列        -->
     <rabbit:queue-arguments>
         <!--3.1 x-dead-letter-exchange：死信交换机名称-->
         <entry key="x-dead-letter-exchange" value="exchange_dlx"/>

         <!--3.2 x-dead-letter-routing-key：从新设置消息的key发送给死信交换机的routingkey-->
         <entry key="x-dead-letter-routing-key" value="dlx.hehe"/>

         <!--4.2 设置队列的长度限制 max-length -->
         <entry key="x-max-length" value="10" value-type="java.lang.Integer"/>
         
      </rabbit:queue-arguments>
 </rabbit:queue>
 
 <!-- 2.正常交换机   -->
<rabbit:topic-exchange name="test_exchange_dlx" id="test_exchange_dlx">
      <rabbit:bindings>
          <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"/>
      </rabbit:bindings>
</rabbit:topic-exchange>


 <!--3.死信队列    -->
<rabbit:queue id="queue_dlx" name="queue_dlx"/>

 <!--4.死信交换机    -->
 <rabbit:topic-exchange name="exchange_dlx" id="exchange_dlx">
      <rabbit:bindings>
          <rabbit:binding pattern="dlx.#" queue="queue_dlx"/>
      </rabbit:bindings>
 </rabbit:topic-exchange>

8.4.4消息拒收

@Component
public class QueueDLX implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收转换消息
            System.out.println(new String(message.getBody()));

            //2. 处理业务逻辑
            System.out.println("处理业务逻辑...");
            int i = 3/0;//出现错误
            //3. 手动签收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            System.out.println("出现异常,拒绝接受");
            //4.拒绝签收,不重回队列 requeue=false
            channel.basicNack(deliveryTag,true,false);   //会进入死信队列
        }
    }
}

8.5 延迟队列

延迟队列存储的对象肯定是对应的延时消息，所谓”延时消息”是指当消息被发送以后，并不想让消费者立即拿到消息，而是等待指定时间后，消费者才拿到这个消息进行消费。
场景：在订单系统中，一个用户下单之后通常有30分钟的时间进行支付，如果30分钟之内没有支付成功，那么这个订单将进行取消处理。这时就可以使用延时队列将订单信息发送到延时队列。
需求：

下单后，30分钟未支付，取消订单，回滚库存。
新用户注册成功30分钟后，发送短信问候。

很可惜，在RabbitMQ中并未提供延迟队列功能。
但是可以使用：TTL+死信队列组合实现延迟队列的效果。

9.SpringBoot集成MQ

9.1整合

9.1.2生产者

<dependencies>
    <!--2. rabbitmq-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
</dependencies>

spring:
  rabbitmq:
    host: 192.168.10.101
    port: 5672
    username: admin
    password: admin
    virtual-host: /

package com.atguigu.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {
    public static final String EXCHANGE_NAME = "boot_topic_exchange";
    public static final String QUEUE_NAME = "boot_queue";


    // 1 交换机
    @Bean("bootExchange")
    public Exchange  bootExchange(){
        return  ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }

    //2.Queue 队列
    @Bean("bootQueue")
    public Queue bootQueue(){
        return  QueueBuilder.durable(QUEUE_NAME).build();
    }

    //3. 队列和交互机绑定关系 Binding
    /*
        1. 知道哪个队列
        2. 知道哪个交换机
        3. routing key
        noargs()：表示不指定参数
     */
    @Bean
    public  Binding bindingQueueExchange(@Qualifier("bootQueue") Queue queue,@Qualifier("bootExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
    }
}

@SpringBootTest
class SpringbootRabbitmqProviderApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() {
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"boot.YXT","mq Springboot");
    }

}

9.1.3消费者

spring:
  rabbitmq:
    host: 192.168.10.101
    port: 5672
    username: admin
    password: admin
    virtual-host: /

@Component
public class RabbimtMQListener {

    @RabbitListener(queues = "boot_queue")  //监听的队列
    public void listenerQueue(Message message){
        System.out.println(message.getMessageProperties().getDeliveryTag());
        System.out.println(new String(message.getBody()));
    }
}

9.1SpringBoot在消费端绑定

** 在消费端绑定交换机，队列，路由key**
** order—->hosp—–>sms**

@Configuration
public class MQConfig {
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

public class MqConst {

    /**
     * 预约下单
     */
    public static final String EXCHANGE_DIRECT_ORDER = "exchange.direct.order";
    public static final String ROUTING_ORDER = "order";
    //队列
    public static final String QUEUE_ORDER  = "queue.order";

    /**
     * 短信
     */
    public static final String EXCHANGE_DIRECT_MSM = "exchange.direct.msm";
    public static final String ROUTING_MSM_ITEM = "msm.item";
    //队列
    public static final String QUEUE_MSM_ITEM  = "queue.msm.item";
}

@Component
public class HospitalReceiver {

    @Autowired
    private ScheduleService scheduleService;

    @Autowired
    private RabbitService rabbitService;


    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = MqConst.QUEUE_ORDER,durable = "true"),
            exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_ORDER),
            key = {MqConst.ROUTING_ORDER}
    ))
    public void receiver(OrderMqVo orderMqVo, Message message, Channel channel) throws IOException {
        Schedule schedule = scheduleService.getById(orderMqVo.getScheduleId());
        schedule.setReservedNumber(orderMqVo.getReservedNumber());
        schedule.setAvailableNumber(orderMqVo.getAvailableNumber());
        scheduleService.update(schedule);
        
        //更新后发送短信告知
        MsmVo msmVo = orderMqVo.getMsmVo();
        if (msmVo!=null){
            rabbitService.sendMessage(MqConst.EXCHANGE_DIRECT_MSM,MqConst.QUEUE_MSM_ITEM,msmVo);
        }
    }
}

/**
 * 消费者端创建交换机，绑定交换机，绑定队列
 */

@Component
public class SmsReceiver {
    @Autowired
    private MsmService msmService;

    @RabbitListener(
    bindings = @QueueBinding(
                value = @Queue(value = MqConst.ROUTING_MSM_ITEM,durable = "true"),
                exchange = @Exchange(value = MqConst.EXCHANGE_DIRECT_MSM),
                key = {MqConst.ROUTING_MSM_ITEM}
                )
         )
    public void send(MsmVo msmVo, Message message, Channel channel) {
        msmService.send(msmVo);
    }
}

@Service
public class RabbitService {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     *  发送消息
     * @param exchange 交换机
     * @param routingKey 路由键
     * @param message 消息
     */
    public boolean sendMessage(String exchange, String routingKey, Object message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
        return true;
    }

}

10.消息百分百成功投递

10.1 生产者端

Step 1：首先把消息信息(业务数据）存储到数据库中，紧接着，我们再把这个消息记录也存储到一张消息记录表里（或者另外一个同源数据库的消息记录表）;
Step 2：发送消息到MQ Broker节点（采用confirm方式发送，会有异步的返回结果）;
Step 3、4：生产者端接受MQ Broker节点返回的Confirm确认消息结果，然后进行更新消息记录表里的消息状态。比如默认Status = 0 当收到消息确认成功后，更新为1即可！
Step 5：但是在消息确认这个过程中可能由于网络闪断、MQ Broker端异常等原因导致回送消息失败或者异常。这个时候就需要发送方（生产者）对消息进行可靠性投递了，保障消息不丢失，100%的投递成功！（有一种极限情况是闪断，Broker返回的成功确认消息，但是生产端由于网络闪断没收到，这个时候重新投递可能会造成消息重复，需要消费端去做幂等处理）所以我们需要有一个定时任务，（比如每5分钟拉取一下处于中间状态的消息，当然这个消息可以设置一个超时时间，比如超过1分钟 Status = 0 ，也就说明了1分钟这个时间窗口内，我们的消息没有被确认，那么会被定时任务拉取出来）
Step 6：接下来我们把中间状态的消息进行重新投递 retry send，继续发送消息到MQ ，当然也可能有多种原因导致发送失败
Step 7：我们可以采用设置最大努力尝试次数，比如投递了3次，**还是失败，那么我们可以将最终状态设置为Status = 2 **，最后交由人工解决处理此类问题（或者把消息转储到失败表中）。

10.2 消费幂等性

消息幂等性保障乐观锁机制
生产者发送消息：
id=1,money=500,version=1
消费者接收消息
id=1,money=500,version=1
id=1,money=500,version=1
消费者需要保证幂等性：第一次执行SQL语句
第一次执行：version=1
update account **set** money = money - 500 , version = version + 1 **where** id = 1 **and** version = 1

消费者需要保证幂等性：第二次执行SQL语句
第二次执行：version=2 (版本已经变更，更新不到了，保证了幂等性)
update account **set** money = money - 500 , version = version + 1 **where** id = 1 **and** version = 1

11.Springboot消息确认

11.1 搭建环境

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


消息持久化
@Queue注解中的属性 - autoDelete：当所有消费客户端连接断开后，是否自动删除队列 。true：删除 false：不删除
@Exchange注解中的属性 - autoDelete：当交换器所有的绑定队列都不再使用时，是否自动删除交换器（更粗粒度，不建议）。true：删除 false：不删除

<dependencies>
    <!--rabbitmq消息队列-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <!--rabbitmq 协议-->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-bus-amqp</artifactId>
    </dependency>
</dependencies>

11.2 生产端消息确认配置1

package com.atguigu.gmall.common.model;
import lombok.Data;
import org.springframework.amqp.rabbit.connection.CorrelationData;

@Data
public class GmallCorrelationData extends CorrelationData {
    //  消息主体
    private Object message;
    //  交换机
    private String exchange;
    //  路由键
    private String routingKey;
    //  重试次数
    private int retryCount = 0;
    //  消息类型  是否是延迟消息
    private boolean isDelay = false;
    //  延迟时间
    private int delayTime = 10;
}

mport org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @Description 消息发送确认
 * <p>
 * ConfirmCallback  只确认消息是否正确到达 Exchange 中
 * ReturnCallback   消息没有正确到达队列时触发回调，如果正确到达队列不执行
 * <p>
 * 1. 如果消息没有到exchange,则confirm回调,ack=false
 * 2. 如果消息到达exchange,则confirm回调,ack=true
 * 3. exchange到queue成功,则不回调return
 * 4. exchange到queue失败,则回调return
 *
 */
@Component
@Slf4j
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);            //指定 ConfirmCallback
        rabbitTemplate.setReturnCallback(this);             //指定 ReturnCallback
    }


    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack){

            log.info("消息发送成功:"+ JSON.toJSONString(correlationData));
        }else {
             log.info("消息发送失败："+cause+"数据"+JSON.toJSONString(correlationData) );
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        //1.反序列化对象输出
        System.out.println("消息主体: " + new String(message.getBody()));
        System.out.println("应答码:"+replyCode);
        System.out.println("描述：" + replyText);
        System.out.println("消息使用的交换器 exchange : " + exchange);
        System.out.println("消息使用的路由键 routing : " + routingKey);
    }
    /**
     * 消息主体: 生产者:发送消息...........
     * 应答码:312
     * 描述：NO_ROUTE
     * 消息使用的交换器 exchange : exchange.confirm
     * 消息使用的路由键 routing : routing.confirm.error
     */
}

11.3生产端消息确认配置:（百分百发送）

  ①在发送端，封装要发送的信息，存入redis
  ②重试的时候获取 GmallCorrelationData类,
  ③重新发送

package com.atguigu.gmall.common.config;
import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.model.GmallCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
@Component
@Slf4j
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {
@Autowired
private  RabbitTemplate rabbitTemplate;
        @PostConstruct
        public void init(){
            
        this.rabbitTemplate.setConfirmCallback(this);
        this.rabbitTemplate.setReturnCallback(this);
    }
    /**
     *  确认消息是否到达交换机
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        if (ack) {
            log.info("消息发送成功：" + JSON.toJSONString(correlationData));
        } else {
            log.info("消息发送失败：" + cause + " 数据：" + JSON.toJSONString(correlationData));

            this.retryMessage(correlationData);
        }
    }

    @Autowired
    private RedisTemplate redisTemplate;
    /**
     * 确实消息是否正确到达队列
     * 执行时机：
     *  到达不执行
     *  没到执行
     *
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message,int replyCode, String replyText, String exchange, String routingKey) {

        System.out.println("消息主体: " + new String(message.getBody()));
        System.out.println("应答码: " + replyCode);
        System.out.println("描述：" + replyText);
        System.out.println("消息使用的交换器 exchange : " + exchange);
        System.out.println("消息使用的路由键 routing : " + routingKey);

        //从redis中获取数据
        String correlationDataId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
        //从redis中获取数据
        String strJson = (String) redisTemplate.opsForValue().get(correlationDataId);
        GmallCorrelationData gmallCorrelationData= JSON.parseObject(strJson, GmallCorrelationData.class);

        //调用重试方法
        this.retryMessage(gmallCorrelationData);
    }

    /**
     * 实现重试方法
     * @param correlationData
     */
    private void retryMessage(CorrelationData correlationData) {

        //转换
        GmallCorrelationData gmallCorrelationData= (GmallCorrelationData) correlationData;
        //获取重试次数
        int retryCount = gmallCorrelationData.getRetryCount();
        //判断
        if(retryCount>3){

            System.out.println("重试次数已到，结束吧");
        }else{
            retryCount+=1;
            gmallCorrelationData.setRetryCount(retryCount);
            //更新redis
            redisTemplate.opsForValue().set(gmallCorrelationData.getId(), JSON.toJSONString(gmallCorrelationData));
            System.out.println("重试次数：\t"+retryCount);
            //重新发送
            this.rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),
                    gmallCorrelationData.getMessage(),gmallCorrelationData);

        }


    }

}

@Service
public class RabbitService {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * 发送消息封装
     * @param exchange
     * @param routingKey
     * @param massage
     * @return
     */
    public boolean sendMessage(String exchange,String routingKey,Object massage){

        //创建实体类封装消息信息
        GmallCorrelationData gmallCorrelationData=new GmallCorrelationData();
        //设置id
        String correlationDataId = UUID.randomUUID().toString().replaceAll("-", "");
       
        gmallCorrelationData.setId(correlationDataId);
        //设置消息
        gmallCorrelationData.setMessage(massage);
        //设置交换机
        gmallCorrelationData.setExchange(exchange);
        //设置路由
        gmallCorrelationData.setRoutingKey(routingKey);
        //存储到redis
        redisTemplate.opsForValue().set(correlationDataId, JSON.toJSONString(gmallCorrelationData),10, TimeUnit.MINUTES);
        //发送消息
//        rabbitTemplate.convertAndSend(exchange,routingKey,massage);
        rabbitTemplate.convertAndSend(exchange,routingKey,massage,gmallCorrelationData);

        return true;
    }

11.4发送消息

@RequestMapping("/mq")
public class MqController {

    @Autowired
    private RabbitService rabbitService;

    /**
     * 消息发送
     */
    //http://localhost:8282/mq/sendConfirm
    @GetMapping("sendConfirm")
    public Result sendConfirm() {
        //rabbitService.sendMessage("exchange.confirm","routing.confirm","生产者:发送消息...........");
        //1.交换机错误
        //rabbitService.sendMessage("exchange.confirm。error","routing.confirm","生产者:发送消息...........");
        //2.队列错误
        rabbitService.sendMessage("exchange.confirm","routing.confirm.error","生产者:发送消息...........");
        return Result.ok();
    }

11.5接收消息

import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;


@Component
public class ConfirmReceiver {


    @SneakyThrows
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "queue.confirm",autoDelete = "false"),
            exchange = @Exchange(value = "exchange.confirm",autoDelete = "true"),
            key = {"routing.confirm"}
    ))
    public void process(Message message, Channel channel){
        //1.接受消息
        System.out.println("监听器:"+new String(message.getBody()));
        System.out.println("接受消息成功...................");
        //2.手动确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

}

12.Springboot延迟队列

12.1 基于死信实现延迟消息

12.1.1消息的TTL（Time To Live）

消息的TTL就是消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL。对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。
如何设置TTL：我们创建一个队列queue.temp，在Arguments 中添加x-message-ttl 为5000 （单位是毫秒），那所有压在这个队列的消息在5秒后会消失。

12.1.2死信交换机 Dead Letter Exchanges

一个消息在满足如下条件下，会进死信路由，记住这里是路由而不是队列，一个路由可以对应很多队列。
（1）一个消息被Consumer拒收了，并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其他消费者使用。
（2）上面的消息的TTL到了，消息过期了。
（3）队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上
Dead Letter Exchange其实就是一种普通的exchange，和创建其他exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有消息过期了，会自动触发消息的转发，发送到Dead Letter Exchange中去。

12.1.3 延迟队列的代码实现

@Configuration
public class DeadLetterMqConfig {

    public static final String exchange_dead = "exchange.dead";
    public static final String routing_dead_1 = "routing.dead.1";
    public static final String routing_dead_2 = "routing.dead.2";
    public static final String queue_dead_1 = "queue.dead.1";
    public static final String queue_dead_2 = "queue.dead.2";


    //1.定义交换机
    @Bean
    public DirectExchange exchange(){
        return  new DirectExchange(exchange_dead,true,false,null);
    }

    @Bean
    public Queue queue1(){
        HashMap<String, Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange",exchange_dead);
        map.put("x-dead-letter-routing-key",routing_dead_2);
        // 设置延迟时间
        map.put("x-message-ttl", 10 * 1000);
        return  new Queue(queue_dead_1,true,false,false,map);
    }

    @Bean
    public Binding binding(){
        // 将队列一 通过routing_dead_1 key 绑定到exchange_dead 交换机上
        return BindingBuilder.bind(queue1()).to(exchange()).with(routing_dead_1);
    }



    //这个队列二就是一个普通队列
    @Bean
    public Queue queue2(){
        return new Queue(queue_dead_2,true,false,false,null);
    }

    //设置队列二的绑定规则
    @Bean
    public Binding binding2(){
        // 将队列二通过routing_dead_2 key 绑定到exchange_dead交换机上！
        return BindingBuilder.bind(queue2()).to(exchange()).with(routing_dead_2);
    }

}

 @Autowired
 private RabbitService rabbitService;

 @Autowired
 private RabbitTemplate rabbitTemplate;

@GetMapping("sendDeadLettle")
   public Result sendDeadLettle() {
       SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
       this.rabbitTemplate.convertAndSend(DeadLetterMqConfig.exchange_dead, DeadLetterMqConfig.routing_dead_1, "ok");
       System.out.println(sdf.format(new Date()) + " Delay sent.");
       return Result.ok();
   }

@Component
public class DeadLetterReceiver {

    @RabbitListener(queues = DeadLetterMqConfig.queue_dead_2)
    public void get(String msg) {
        System.out.println("Receive:" + msg);
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("Receive queue_dead_2: " + sdf.format(new Date()) + " Delay rece." + msg);
    }
}


Receive:ok
Receive queue_dead_2: 2023-05-19 12:48:53 Delay rece.ok

12.2基于延迟插件实现延迟消息

Rabbitmq实现了一个插件x-delay-message来实现延时队列

首先我们将刚下载下来的rabbitmq_delayed_message_exchange-3.8.0.ez文件上传到RabbitMQ所在服务器，下载地址：https://www.rabbitmq.com/community-plugins.html
切换到插件所在目录，执行 docker cp rabbitmq_delayed_message_exchange-3.8.0.ez rabbitmq:/plugins 命令，将刚插件拷贝到容器内plugins目录下
执行 docker exec -it rabbitmq /bin/bash 命令进入到容器内部，并 cd plugins 进入plugins目录
执行 ls -l|grep delay 命令查看插件是否copy成功
在容器内plugins目录下，执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 命令启用插件
exit命令退出RabbitMQ容器内部，然后执行 docker restart rabbitmq 命令重启RabbitMQ容器

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class DelayedMqConfig {

    public static final String exchange_delay = "exchange.delay";
    public static final String routing_delay = "routing.delay";
    public static final String queue_delay_1 = "queue.delay.1";


    @Bean
    public Queue delayQeue1() {
        // 第一个参数是创建的queue的名字，第二个参数是是否支持持久化
        return new Queue(queue_delay_1, true);
    }


    //插件方式
    @Bean
    public CustomExchange delayExchange(){
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-delayed-type","direct");
        return new CustomExchange(exchange_delay, "x-delayed-message", true, false, args);
    }


    @Bean
    public Binding delayBbinding1() {
        return BindingBuilder.bind(delayQeue1()).to(delayExchange()).with(routing_delay).noargs();
    }
}

@GetMapping("sendDelay")
public Result sendDelay() {
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    this.rabbitTemplate.convertAndSend(DelayedMqConfig.exchange_delay, DelayedMqConfig.routing_delay, sdf.format(new Date()), new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            message.getMessageProperties().setDelay(10 * 1000);
            System.out.println(sdf.format(new Date()) + " Delay sent.");
            return message;
        }
    });
    return Result.ok();
}

@Component
public class DelayReceiver {

    @RabbitListener(queues = DelayedMqConfig.queue_delay_1)
    public void get(String msg) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("Receive queue_delay_1: " + sdf.format(new Date()) + " Delay rece." + msg);
    }

}

12.2.1幂等性处理

package com.atguigu.gmall.mq.receiver;

import com.atguigu.gmall.mq.config.DelayedMqConfig;
import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;
@Component
public class DelayedReceiver {
    @Autowired
    private RedisTemplate redisTemplate;
    /**
     * 实现幂等性思路：
     *
     *   1.是否接收消息添加标记
     *     reids setNx
     *   2.如果redis中存在了，一定是消费了吗？
     *       未消费 0
     *       消费  1
     * @param msg
     * @param message
     * @param channel
     */
    @RabbitListener(queues = DelayedMqConfig.queue_delay_1)
    @SneakyThrows
    public void getMessage(String msg, Message message, Channel channel){

        String delayKey="delayed:";

        //设置消息到redis --第一次  true
        Boolean result = redisTemplate.opsForValue().setIfAbsent(delayKey + msg, "0", 10, TimeUnit.MINUTES);
        //判断
        if(!result){

            //判断是否消费
            String flag = (String) this.redisTemplate.opsForValue().get(delayKey + msg);
            //判断
            if("1".equals(flag)){
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }else{
                SimpleDateFormat simpleDateFormat=new SimpleDateFormat("yyyy-MM-dd  HH:mm:ss");
                System.out.println("消息接收的时间：\t"+simpleDateFormat.format(new Date()));
                System.out.println("消息的内容是："+msg);
                //设置消息标识为1
                this.redisTemplate.opsForValue().set(delayKey + msg,"1",10,TimeUnit.MINUTES);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }
        }else{
            SimpleDateFormat simpleDateFormat=new SimpleDateFormat("yyyy-MM-dd  HH:mm:ss");
            System.out.println("消息接收的时间：\t"+simpleDateFormat.format(new Date()));
            System.out.println("消息的内容是："+msg);
            //设置消息标识为1
            this.redisTemplate.opsForValue().set(delayKey + msg,"1",10,TimeUnit.MINUTES);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
    }
}

12.2.2 重试机制

package com.atguigu.gmall.common.config;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.model.GmallCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;

@Component
@Slf4j
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {

@Autowired
private  RabbitTemplate rabbitTemplate;

        @PostConstruct
        public void init(){

        this.rabbitTemplate.setConfirmCallback(this);
        this.rabbitTemplate.setReturnCallback(this);


    }
    /**
     *  确认消息是否到达交换机
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {


        if (ack) {
            log.info("消息发送成功：" + JSON.toJSONString(correlationData));
        } else {
            log.info("消息发送失败：" + cause + " 数据：" + JSON.toJSONString(correlationData));

            this.retryMessage(correlationData);
        }


    }

    @Autowired
    private RedisTemplate redisTemplate;


    /**
     * 确实消息是否正确到达队列
     * 执行时机：
     *  到达不执行
     *  没到执行
     *
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message,int replyCode, String replyText, String exchange, String routingKey) {

        System.out.println("消息主体: " + new String(message.getBody()));
        System.out.println("应答码: " + replyCode);
        System.out.println("描述：" + replyText);
        System.out.println("消息使用的交换器 exchange : " + exchange);
        System.out.println("消息使用的路由键 routing : " + routingKey);

        //从redis中获取数据
        String correlationDataId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
        if(!StringUtils.isEmpty(correlationDataId)){

            //从redis中获取数据
            String strJson = (String) redisTemplate.opsForValue().get(correlationDataId);

            GmallCorrelationData gmallCorrelationData= JSON.parseObject(strJson, GmallCorrelationData.class);

            //调用重试方法
            this.retryMessage(gmallCorrelationData);

        }



    }

    /**
     * 实现重试方法
     * @param correlationData
     */
    private void retryMessage(CorrelationData correlationData) {

        //转换
        GmallCorrelationData gmallCorrelationData= (GmallCorrelationData) correlationData;
        //获取重试次数
        int retryCount = gmallCorrelationData.getRetryCount();
        //判断
        if(retryCount>3){

            System.out.println("重试次数已到，结束吧");
        }else{
            retryCount+=1;
            gmallCorrelationData.setRetryCount(retryCount);
            //更新redis
            redisTemplate.opsForValue().set(gmallCorrelationData.getId(), JSON.toJSONString(gmallCorrelationData));
            System.out.println("重试次数：\t"+retryCount);
            //判断是否延迟
            if(gmallCorrelationData.isDelay()){

                this.rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),
                        gmallCorrelationData.getMessage(),message -> {

                            message.getMessageProperties().setDelay(gmallCorrelationData.getDelayTime()*1000);
                        return message;
                        },gmallCorrelationData);

            }else{

                //重新发送
                this.rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),
                        gmallCorrelationData.getMessage(),gmallCorrelationData);

            }
        }
    }
}

12.2.3 延迟消息发送封装

@Service
public class RabbitService {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     *
     * @param exchange
     * @param routingKey
     * @param meggage
     * @return
     */
    public  boolean sendMessage(String exchange,String routingKey,Object meggage){
        rabbitTemplate.convertAndSend(exchange,routingKey,meggage);
        return true;
    }


    /**
     * 发送延迟消息
     * @param exchange 交换机
     * @param routingKey 路由键
     * @param message 消息
     * @param delayTime 单位：秒
     */
    public boolean sendDelayMessage(String exchange, String routingKey, Object message, int delayTime) {
          this.rabbitTemplate.convertAndSend(exchange,routingKey,message,msg -> {
              //设置消息延迟时间
              msg.getMessageProperties().setDelay(delayTime*1000);
              return  msg;
          });
          return  true;
    }
}

@Configuration
public class OrderCanelMqConfig {

    /**
     * 队列
     * @return
     */
    @Bean
    public Queue delayQueue(){
        return  new Queue(MqConst.QUEUE_ORDER_CANCEL,true);
    }


    /**
     * 延迟交换机
     * @return
     */
    @Bean
    public CustomExchange delayExchange() {
        HashMap<String, Object> map = new HashMap<>();
        map.put("x-delayed-type","direct");
        return new CustomExchange(MqConst.EXCHANGE_DIRECT_ORDER_CANCEL,"x-delayed-message",true,false,map);
    }


    /**
     * 绑定队列与交换机
     * @return
     */
    @Bean
    public Binding bindingDelay() {
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(MqConst.ROUTING_ORDER_CANCEL).noargs();
    }


}

12.3超时订单关闭

package com.atguigu.gmall.order.config;
import com.atguigu.gmall.common.constant.MqConst;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;

@Configuration
public class OrderCanelMqConfig {

    /**
     * 队列
     * @return
     */
    @Bean
    public Queue delayQueue(){
        return  new Queue(MqConst.QUEUE_ORDER_CANCEL,true);
    }


    /**
     * 延迟交换机
     * @return
     */
    @Bean
    public CustomExchange delayExchange() {
        HashMap<String, Object> map = new HashMap<>();
        map.put("x-delayed-type","direct");
        return new CustomExchange(MqConst.EXCHANGE_DIRECT_ORDER_CANCEL,"x-delayed-message",true,false,map);
    }


    /**
     * 绑定队列与交换机
     * @return
     */
    @Bean
    public Binding bindingDelay() {
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(MqConst.ROUTING_ORDER_CANCEL).noargs();
    }


}

@Override
@Transactional(rollbackFor = Exception.class)
public Long saveOrderInfo(OrderInfo orderInfo) {
        orderInfo.sumTotalAmount();
        //1.设置支付状态
        orderInfo.setOrderStatus(OrderStatus.UNPAID.name());
        //2.设置订单交易号
        String outTradeNo = "ATGUIGU" + System.currentTimeMillis() + "" + new Random().nextInt(1000);
        orderInfo.setOutTradeNo(outTradeNo);
        //3.设置创建时间
        orderInfo.setCreateTime(new Date());
        //4.设置过期时间
        Calendar calendar = Calendar.getInstance();
        //日期加一天
        calendar.add(Calendar.DATE,1);
        orderInfo.setExpireTime(calendar.getTime());
        //5.设置支付处理状态
        orderInfo.setProcessStatus(ProcessStatus.UNPAID.name());
        //6.获取订单明细
        List<OrderDetail> orderDetailList = orderInfo.getOrderDetailList();
        StringBuffer tradeBody = new StringBuffer();
        for (OrderDetail orderDetail : orderDetailList) {
            tradeBody.append(orderDetail.getSkuName()+" ");
        }
        if (tradeBody.toString().length()>100){
            orderInfo.setTradeBody(tradeBody.toString().substring(0,100));
        }else {
            orderInfo.setTradeBody(tradeBody.toString());
        }

        //7.保存订单信息
        orderInfoMapper.insert(orderInfo);

        //8.保存订单详情
        for (OrderDetail orderDetail : orderDetailList) {
            orderDetail.setOrderId(orderInfo.getId());
            orderDetailMapper.insert(orderDetail);
        }

        //发送延迟队列，如果定时未支付，取消订单
        rabbitService.sendDelayMessage(MqConst.EXCHANGE_DIRECT_ORDER_CANCEL, MqConst.ROUTING_ORDER_CANCEL, orderInfo.getId(), MqConst.DELAY_TIME);

        //9.返回订单Id
        return orderInfo.getId();
}

package com.atguigu.gmall.order.receiver;

import com.atguigu.gmall.model.order.OrderInfo;
import com.atguigu.gmall.order.service.OrderService;
import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class OrderReceiver {
    @Autowired
    private OrderService orderService;


    @SneakyThrows
    @RabbitListener(queues = MqConst.QUEUE_ORDER_CANCEL)
    public void cancelOrder(Long orderId , Message message, Channel channel){
        try{
             if (orderId!=null){
                 //  发过来的是订单Id，那么你就需要判断一下当前的订单是否已经支付了。
                 //  未支付的情况下：关闭订单
                 //  根据订单Id 查询orderInfo select * from order_info where id = orderId
                 //  利用这个接口IService  实现类ServiceImpl 完成根据订单Id 查询订单信息 ServiceImpl 类底层还是使用的mapper
                 OrderInfo orderInfo = orderService.getById(orderId);
                 if (orderInfo!=null&&"UNPAID".equals(orderInfo.getOrderStatus())&&"UNPAID".equals(orderInfo.getProcessStatus())){
                        //关闭订单
                     orderService.execExpiredOrder(orderId);
                 }
             }

        }catch (Exception e){
            e.printStackTrace();
        }
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

    }

}

@Override
public void execExpiredOrder(Long orderId) {
    // orderInfo
    updateOrderStatus(orderId, ProcessStatus.CLOSED);
}

@Override
public void updateOrderStatus(Long orderId, ProcessStatus processStatus) {
    OrderInfo orderInfo = new OrderInfo();
    orderInfo.setId(orderId);
    orderInfo.setProcessStatus(processStatus.name());
    orderInfo.setOrderStatus(processStatus.getOrderStatus().name());
    orderInfoMapper.updateById(orderInfo);
}