深入浅出RabbitMQ

reading: 2024-02-05

1、中间件

中间件（Middleware）是介于应用系统和系统软件之间的一类软件，它使用系统软件所提供的基础服务（功能），衔接网络上应用系统的各个部分或不同的应用，能够达到资源共享、功能共享的目的。它并没有很严格的定义，但是普遍接受IDC的定义：中间件是一种独立的系统软件服务程序，分布式应用软件借助这种软件在不同的技术之间共享资源，中间件位于客户机服务器的操作系统之上，管理计算资源和网络通信。从这个意义上可以用一个等式来表示中间件：中间件=平台+通信，这也就限定了只有用于分布式系统中才能叫中间件，同时也把它与支撑软件和实用软件区分开来。

1.1、常用中间件

分布式消息中间件：

ActiveMQ
RabbitMQ
Kafka
RocketMQ

使用场景：

消息中间件监控数据
异步数据传输
削峰填谷场景
任务调度场景
海量数据同步场景
分布式事务场景
日志管理场景
大数据分析场景

常见协议：

AMQP
MQTT
Kafka

负载均衡中间件：

Nginx
LVS
KeepAlive
CDN

缓存中间件：

Memcache
Redis

数据库中间件：

Mycat
ShardingJdbc

1.2、基于消息中间件的分布式系统的架构

消息中间件的本质 是一种接受数据，接受请求、存储数据、发送数据等功能的技术服务。

2、消息队列协议

2.1、协议

网络协议通常包含三个基本要素：

语法：它定义了用户数据与控制信息的结构及格式。
语义：这涉及到需要发出的控制信息及其含义，以及完成这些动作后的响应和行为。
时序：它描述了事件的实现顺序和时间上的先后关系，确保数据的正确排序和传输。

这三个要素共同构成了网络协议的核心，确保通信双方能够有效地交换信息和管理通信过程。

为什么中间件不直接使用HTTP协议？

因为http请求报文头和响应报文头是比较复杂的，包含了cookie，数据的加密解密，状态码，响应码等附加的功能，但是对于一个消息而言，我们并不需要这么复杂，也没有这个必要性，它其实就是负责把数据传递、存储、分发就行，一定要追求的是高性能，尽量简洁、快速。
大部分情况下http大部分都是短链接，在实际的交互过程中，一个请求到响应很有可能会中断，中断以后就不会进行持久化，就会造成请求的丢失，这样就不利于消息中间件的业务场景，因为消息中间件可能是一个长期的获取消息的过程，出现问题和故障要对数据或消息进行持久化等，目的是为了保证消息和数据的高可靠和稳键的运行。

2.2、常见消息队列协议

AMQP

AMQP，即Advanced Message Queuing Protocol（高级消息队列协议），一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件传递消息，不受客户端/中间件不同产品、不同开发语言等条件的限制。该协议是一种二进制协议，提供客户端应用于消息中间件之间异步、安全、高效的交互。相对于我们常见的REST API，AMQP更容易实现，可以降低开销，同时灵活性高，可以轻松的添加负载平衡和高可用性的功能，并保证消息传递，在性能上AMQP协议也相对更好一些。

特性:

分布式事务支持。
消息的持久化支持。
高性能和高可靠的消息处理优势。

MQTT

MQTT(消息队列遥测传输)是ISO标准(ISO/IEC PRF 20922)下基于发布订阅范式的消息协议。它工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，为此，它需要一个消息中间件。

特点：

轻量
结构简单
传输快，不支持事务
没有持久化设计

应用场景：

适用于计算能力有限
低带宽
网络不稳定的场景

OpenMessage

是近几年由阿里、雅虎和滴滴出行、Stremalio等公司共同参与创立的分布式信息中间件、流处理等领域的应用开发标准

特点：

结构简单
解析速度快
支持事务和持久化设计

Kafka

Kafka协议是基于 TCP/IP的二进制协议。消息内部是通过长度来分割，由一些基本数据类型组成。

特点：

结构简单
解析速度快
无事务支持
有持久化设计

3、消息队列对比

将数据存入磁盘，而不是存在内存中随服务器重启断开而消失，使数据能够永久保存。

消息队列持久化：

	ActiveMQ	RabbitMQ	Kafka	RocketMQ
文件存储	支持	支持	支持	支持
数据库	支持	/	/	/

消息分发策略的机制：

	ActiveMQ	RabbitMQ	Kafka	RocketMQ
发布订阅	支持	支持	支持	支持
轮询分发	支持	支持	支持	/
公平分发	/	支持	支持	/
重发	支持	支持	/	支持
消息拉取	/	支持	支持	支持

4、高可用与高可靠

4.1、高可用

什么是高可用？

高可用（High availability,缩写为 HA)，是指系统无中断地执行其功能的能力，代表系统的可用性程度。高可用的主要目的是为了保障“业务的连续性”，即在用户眼里，业务永远是正常对外提供服务的。

高可用级别

可用性级别	系统可用性%	年宕机时间	月宕机时间	周宕机时间	每天宕机时间
不可用	90%	36.5天	73小时	16.8小时	144分钟
基本可用	99%	87.6小时	7.3小时	1.68小时	14.4分钟
较高可用	99.9%	8.76小时	43.8分钟	10.1分钟	1.44分钟
高可用	99.99%	52.56分钟	4.38分钟	1.01秒	8.64秒
极高可用	99.999%	5.26分钟	26.28秒	6.06秒	0.86秒

4.1.1、集群模式

主从共享数据

生产者将消息发送到Master节点，所有的都连接这个消息队列共享这块数据区域，Master节点负责写入，一旦Master挂掉，slave节点继续服务。从而形成高可用。

主从同步部署

这种模式写入消息同样在Master主节点上，但是主节点会同步数据到slave节点形成副本，和zookeeper或者redis主从机制很类同。这样可以达到负载均衡的效果，如果消费者有多个这样就可以去不同的节点就行消费，以为消息的拷贝和同步会暂用很大的带宽和网络资源。在后续的rabbtmq中会有使用。

多主机群同步部署

和上面的区别不是特别的大，但是它的写入可以往任意节点去写入。

多主集群转发部署

如果你插入的数据是broker-1中，元数据信息会存储数据的相关描述和记录存放的位置（队列）。
它会对描述信息也就是元数据信息就行同步，如果消费者在broker-2中进行消费，发现自己几点没有对应的消息，可以从对应的元数据信息中去查询，然后返回对应的消息信息，场景：比如买火车票或者黄牛买演唱会门票，比如第一个黄牛有顾客说要买的演唱会门票，但是没有但是他会去联系其他的黄牛询问，如果有就返回。

Master-slave与roker-cluster组合

实现多主多从的热备机制来完成消息的高可用以及数据的热备机制，在生产规模达到一定的阶段的时候，这种使用的频率比较高。

总结：
1：要么消息共享，
2：要么消息同步
3：要么元数据共享

4.2、高可靠

定义：是指系统可以无故障低持续运行，比如一个系统突然崩溃，报错，异常等等并不影响线上业务的正常运行，出错的几率极低，就称之为：高可靠。

两个方面考虑：
1：消息的传输：通过协议来保证系统间数据解析的正确性。
2：消息的存储可靠：通过持久化来保证消息的可靠性。

5、安装

rabbitmq官网：https://www.rabbitmq.com/

下载地址：https://www.rabbitmq.com/download.html

RabbitMQ是一个开源的遵循AMQP协议实现的基于Erlang语言编写，支持多种客户端（语言）。用于在分布式系统中存储消息，转发消息，具有高可用，高可扩性，易用性等特征。

Erlang限制

RabbitMQ Version	Minimum Erlang/OTP	Maximum Erlang/OTP	Erlang/OTP Notes
3.12.12	25.0	26.2.x	Compatible with Erlang 26
3.12.11	25.0	26.2.x	Compatible with Erlang 26
3.12.10	25.0	26.2.x	Compatible with Erlang 26
…	…	…	…
3.11.0	25.0	25.3.x	Erlang 26 supported starting with RabbitMQ 3.12.0

5.1、Linux安装

较旧的发行版也可能缺少足够新的 OpenSSL 版本。Erlang 24不能在不提供 OpenSSL 1.1作为系统库的发行版上使用。CentOS 7 和 Fedora 26 之前的版本就是此类发行版的示例。

目前受支持的基于 RPM 的发行版列表包括

Fedora 35 through 39
CentOS Stream 9.x and 8.x
RedHat Enterprise Linux 9.x and 8.x
Amazon Linux 2023
Rocky Linux 9.x and 8.x
Alma Linux 9.x and 8.x
Oracle Linux 9.x and 8.x
openSUSE Leap 15.3 and later versions

如果满足依赖性，这些软件包可以在其他基于 RPM 的发行版上运行，但它们的测试和支持是在尽最大努力的基础上完成的。

因此在centos7.x安装rabbitmq需要更新openssl

安装erlang：

1 2	yum install -y erlang erl -v

安装socat:

1	yum install -y socat

下载安装：

1 2	> wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.13/rabbitmq-server-3.8.13-1.el8.noarch.rpm > rpm -Uvh rabbitmq-server-3.8.13-1.el8.noarch.rpm

常用服务命令：

# 启动服务
> systemctl start rabbitmq-server
# 查看服务状态
> systemctl status rabbitmq-server
# 停止服务
> systemctl stop rabbitmq-server
# 开机启动服务
> systemctl enable rabbitmq-server

RabbitMQ默认情况下有一个配置文件，定义了RabbitMQ的相关配置信息，默认情况下能够满足日常的开发需求。如果需要修改需要，需要自己创建一个配置文件进行覆盖。

文档：https://www.rabbitmq.com/documentation.html

端口：

5672:RabbitMQ的通讯端口
25672:RabbitMQ的节点间的CLI通讯端口是
15672:RabbitMQ HTTP_API的端口，管理员用户才能访问，用于管理RabbitMQ,需要启动Management插件。
1883，8883：MQTT插件启动时的端口。
61613、61614：STOMP客户端插件启用的时候的端口。
15674、15675：基于webscoket的STOMP端口和MOTT端口

开放15672的端口。

5.2、Docker安装

安装docker

（1）yum 包更新到最新
> yum update
（2）安装需要的软件包， yum-util 提供yum-config-manager功能，另外两个是devicemapper驱动依赖的
> yum install -y yum-utils device-mapper-persistent-data lvm2
（3）设置yum源为阿里云
> yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo
（4）安装docker
> yum install docker-ce -y
（5）安装后查看docker版本
> docker -v
 (6) 安装加速镜像
 sudo mkdir -p /etc/docker
 sudo tee /etc/docker/daemon.json <<-'EOF'
 {
  "registry-mirrors": ["https://0wrdwnn6.mirror.aliyuncs.com"]
 }
 EOF
 sudo systemctl daemon-reload
 sudo systemctl restart docker

docker相关命令

# 启动docker：
systemctl start docker
# 停止docker：
systemctl stop docker
# 重启docker：
systemctl restart docker
# 查看docker状态：
systemctl status docker
# 开机启动：  
systemctl enable docker
systemctl unenable docker
# 查看docker概要信息
docker info
# 查看docker帮助文档
docker --help

获取rabbit镜像

1	docker pull rabbitmq:management

创建并运行容器

1	docker run -di --name=myrabbit -p 15672:15672 rabbitmq:management

—hostname：指定容器主机名称
—name:指定容器名称
-p:将mq端口号映射到本地

运行时设置用户和密码

1	docker run -di --name myrabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 -p 25672:25672 -p 61613:61613 -p 1883:1883 rabbitmq:management

5.3、RabbitMQ常用命令

新增用户

1	rabbitmqctl add_user admin admin

设置用户分配操作权限

1	rabbitmqctl set_user_tags admin administrator

用户级别：

administrator 可以登录控制台、查看所有信息、可以对rabbitmq进行管理
monitoring 监控者登录控制台，查看所有信息
policymaker 策略制定者登录控制台,指定策略
managment 普通管理员登录控制台

为用户添加资源权限

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

总结：

rabbitmqctl add_user 账号 密码
rabbitmqctl set_user_tags 账号 administrator
rabbitmqctl change_password Username Newpassword 修改密码
rabbitmqctl delete_user Username 删除用户
rabbitmqctl list_users 查看用户清单
rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*" 为用户设置administrator角色
rabbitmqctl set_permissions -p / root ".*" ".*" ".*"

6、角色分类

none：

不能访问management plugin

management：查看自己相关节点信息

列出自己可以通过AMQP登入的虚拟机
查看自己的虚拟机节点 virtual hosts的queues,exchanges和bindings信息
查看和关闭自己的channels和connections
查看有关自己的虚拟机节点virtual hosts的统计信息。包括其他用户在这个节点virtual hosts中的活动信息。

Policymaker

包含management所有权限
查看和创建和删除自己的virtual hosts所属的policies和parameters信息。

Monitoring

包含management所有权限
罗列出所有的virtual hosts，包括不能登录的virtual hosts。
查看其他用户的connections和channels信息
查看节点级别的数据如clustering和memory使用情况
查看所有的virtual hosts的全局统计信息。

Administrator

最高权限
可以创建和删除virtual hosts
可以查看，创建和删除users
查看创建permisssions
关闭所有用户的connections

7、核心组成

核心概念：
Server：又称Broker ,接受客户端的连接，实现AMQP实体服务。安装rabbitmq-server
Connection：连接，应用程序与Broker的网络连接 TCP/IP/ 三次握手和四次挥手
Channel：网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道，客户端可以建立对各Channel，每个Channel代表一个会话任务。
Message :消息：服务与应用程序之间传送的数据，由Properties和body组成，Properties可是对消息进行修饰，比如消息的优先级，延迟等高级特性，Body则就是消息体的内容。
Virtual Host 虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个虚拟主机理由可以有若干个Exhange和Queueu，同一个虚拟主机里面不能有相同名字的Exchange
Exchange：交换机，接受消息，根据路由键发送消息到绑定的队列。(==不具备消息存储的能力==)
Bindings：Exchange和Queue之间的虚拟连接，binding中可以保护多个routing key.
Routing key：是一个路由规则，虚拟机可以用它来确定如何路由一个特定消息。
Queue：队列：也成为Message Queue,消息队列，保存消息并将它们转发给消费者。

8、七种工作模式

官网文档：https://www.rabbitmq.com/getstarted.html

8.1、simple

也称为队列模型或任务模型。
消息生产者发送消息到队列，而消息消费者从队列中接收并处理消息。
消息被消费后就从队列中删除。

public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("47.104.141.27");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4: 从连接中获取通道channel
            channel = connection.createChannel();
            // 5: 申明队列queue存储消息
            /*
             *  如果队列不存在，则会创建
             *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
             *
             *  @params1： queue 队列的名称
             *  @params2： durable 队列是否持久化
             *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
             *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
             *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
             * */
            channel.queueDeclare("queue1", false, false, false, null);
            // 6： 准备发送消息的内容
            String message = "你好，学相伴！！！";
            // 7: 发送消息给中间件rabbitmq-server
            // @params1: 交换机exchange
            // @params2: 队列名称/routing
            // @params3: 属性配置
            // @params4: 发送消息的内容
            channel.basicPublish("", "queue1", null, message.getBytes());
            System.out.println("消息发送成功!");
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("发送消息出现异常...");
        } finally {
            // 7: 释放连接关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }

8.2、work

工作队列模式（Work Queues）：

也称为任务队列模式。
多个消费者共享一个队列，但每个消息只能被一个消费者处理。
消息在队列中的消费顺序是按照先到先服务（FIFO）的方式。

发放策略：

轮询分发方式
公平分发方式

发布/订阅模式（Publish/Subscribe）：

允许一个消息被多个消费者接收。
消息生产者将消息发布到交换机（exchange），而队列通过绑定到交换机的方式接收消息。
一个消息被发布到交换机后，所有绑定到该交换机的队列都会收到这个消息的副本。

实现：

类型：fanout
特点：Fanout—发布与订阅模式，是一种广播机制，它是没有路由key的模式。

public static void main(String[] args) {
       // 1: 创建连接工厂
       ConnectionFactory connectionFactory = new ConnectionFactory();
       // 2: 设置连接属性
       connectionFactory.setHost("47.104.141.27");
       connectionFactory.setPort(5672);
       connectionFactory.setVirtualHost("/");
       connectionFactory.setUsername("admin");
       connectionFactory.setPassword("admin");
       Connection connection = null;
       Channel channel = null;
       try {
           // 3: 从连接工厂中获取连接
           connection = connectionFactory.newConnection("生产者");
           // 4: 从连接中获取通道channel
           channel = connection.createChannel();
           // 6： 准备发送消息的内容
           String message = "你好，学相伴！！！";
           String  exchangeName = "fanout-exchange";
           String routingKey = "";
           // 7: 发送消息给中间件rabbitmq-server
           // @params1: 交换机exchange
           // @params2: 队列名称/routingkey
           // @params3: 属性配置
           // @params4: 发送消息的内容
           channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
           System.out.println("消息发送成功!");
       } catch (Exception ex) {
           ex.printStackTrace();
           System.out.println("发送消息出现异常...");
       } finally {
           // 7: 释放连接关闭通道
           if (channel != null && channel.isOpen()) {
               try {
                   channel.close();
               } catch (Exception ex) {
                   ex.printStackTrace();
               }
           }
           if (connection != null) {
               try {
                   connection.close();
               } catch (Exception ex) {
                   ex.printStackTrace();
               }
           }
       }
   }

消费者：

private static Runnable runnable = () -> {
    // 1: 创建连接工厂
    ConnectionFactory connectionFactory = new ConnectionFactory();
    // 2: 设置连接属性
    connectionFactory.setHost("47.104.141.27");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
    connectionFactory.setUsername("admin");
    connectionFactory.setPassword("admin");
    //获取队列的名称
    final String queueName = Thread.currentThread().getName();
    Connection connection = null;
    Channel channel = null;
    try {
        // 3: 从连接工厂中获取连接
        connection = connectionFactory.newConnection("生产者");
        // 4: 从连接中获取通道channel
        channel = connection.createChannel();
        // 5: 申明队列queue存储消息
        /*
         *  如果队列不存在，则会创建
         *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
         *
         *  @params1： queue 队列的名称
         *  @params2： durable 队列是否持久化
         *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
         *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
         *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
         * */
        // 这里如果queue已经被创建过一次了，可以不需要定义
        //channel.queueDeclare("queue1", false, false, false, null);
        // 6： 定义接受消息的回调
        Channel finalChannel = channel;
        finalChannel.basicConsume(queueName, true, new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));
            }
        }, new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {
            }
        });
        System.out.println(queueName + "：开始接受消息");
        System.in.read();
    } catch (Exception ex) {
        ex.printStackTrace();
        System.out.println("发送消息出现异常...");
    } finally {
        // 7: 释放连接关闭通道
        if (channel != null && channel.isOpen()) {
            try {
                channel.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
        if (connection != null && connection.isOpen()) {
            try {
                connection.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }
};
public static void main(String[] args) {
    // 启动三个线程去执行
    new Thread(runnable, "queue-1").start();
    new Thread(runnable, "queue-2").start();
    new Thread(runnable, "queue-3").start();
}

8.4、Routing

实现：

类型：direct
特点：Direct模式是fanout模式上的一种叠加，增加了路由RoutingKey的模式。

生产者：

public class Producer {
    public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("47.104.141.27");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4: 从连接中获取通道channel
            channel = connection.createChannel();
            // 6： 准备发送消息的内容
            String message = "你好，学相伴！！！";
            String  exchangeName = "direct-exchange";
            String routingKey1 = "testkey";
            String routingKey2 = "testkey2";
            // 7: 发送消息给中间件rabbitmq-server
            // @params1: 交换机exchange
            // @params2: 队列名称/routingkey
            // @params3: 属性配置
            // @params4: 发送消息的内容
            channel.basicPublish(exchangeName, routingKey1, null, message.getBytes());
            channel.basicPublish(exchangeName, routingKey2, null, message.getBytes());
            System.out.println("消息发送成功!");
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("发送消息出现异常...");
        } finally {
            // 7: 释放连接关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}

消费者：

public class Consumer {
    private static Runnable runnable = () -> {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("47.104.141.27");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        //获取队列的名称
        final String queueName = Thread.currentThread().getName();
        Connection connection = null;
        Channel channel = null;
        try {
            // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4: 从连接中获取通道channel
            channel = connection.createChannel();
            // 5: 申明队列queue存储消息
            /*
             *  如果队列不存在，则会创建
             *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
             *
             *  @params1： queue 队列的名称
             *  @params2： durable 队列是否持久化
             *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
             *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
             *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
             * */
            // 这里如果queue已经被创建过一次了，可以不需要定义
            //channel.queueDeclare("queue1", false, false, false, null);
            // 6： 定义接受消息的回调
            Channel finalChannel = channel;
            finalChannel.basicConsume(queueName, true, new DeliverCallback() {
                @Override
                public void handle(String s, Delivery delivery) throws IOException {
                    System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));
                }
            }, new CancelCallback() {
                @Override
                public void handle(String s) throws IOException {
                }
            });
            System.out.println(queueName + "：开始接受消息");
            System.in.read();
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("发送消息出现异常...");
        } finally {
            // 7: 释放连接关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            if (connection != null && connection.isOpen()) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    };
    public static void main(String[] args) {
        // 启动三个线程去执行
        new Thread(runnable, "queue-1").start();
        new Thread(runnable, "queue-2").start();
        new Thread(runnable, "queue-3").start();
    }
}

8.5、Topics

主题模式（Topic）：

是发布/订阅模式的扩展，允许消息的选择性接收。
在主题模式中，队列可以使用通配符（例如，*和#）绑定到交换机，以指定它们感兴趣的消息。
通配符 * 表示一个单词，而 # 表示零个或多个单词。

实现：

类型：topic
特点：Topic模式是direct模式上的一种叠加，增加了模糊路由RoutingKey的模式。

public class Producer {
    public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("47.104.141.27");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4: 从连接中获取通道channel
            channel = connection.createChannel();
            // 6： 准备发送消息的内容
            String message = "你好，学相伴！！！";
            String  exchangeName = "topic-exchange";
            String routingKey1 = "com.course.order";//都可以收到 queue-1 queue-2
            String routingKey2 = "com.order.user";//都可以收到 queue-1 queue-3
            // 7: 发送消息给中间件rabbitmq-server
            // @params1: 交换机exchange
            // @params2: 队列名称/routingkey
            // @params3: 属性配置
            // @params4: 发送消息的内容
            channel.basicPublish(exchangeName, routingKey1, null, message.getBytes());
            System.out.println("消息发送成功!");
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("发送消息出现异常...");
        } finally {
            // 7: 释放连接关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}

消费者：

public class Consumer {
    private static Runnable runnable = () -> {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("47.104.141.27");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        //获取队列的名称
        final String queueName = Thread.currentThread().getName();
        Connection connection = null;
        Channel channel = null;
        try {
            // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4: 从连接中获取通道channel
            channel = connection.createChannel();
            // 5: 申明队列queue存储消息
            /*
             *  如果队列不存在，则会创建
             *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
             *
             *  @params1： queue 队列的名称
             *  @params2： durable 队列是否持久化
             *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
             *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
             *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
             * */
            // 这里如果queue已经被创建过一次了，可以不需要定义
            //channel.queueDeclare("queue1", false, false, false, null);
            // 6： 定义接受消息的回调
            Channel finalChannel = channel;
            finalChannel.basicConsume(queueName, true, new DeliverCallback() {
                @Override
                public void handle(String s, Delivery delivery) throws IOException {
                    System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));
                }
            }, new CancelCallback() {
                @Override
                public void handle(String s) throws IOException {
                }
            });
            System.out.println(queueName + "：开始接受消息");
            System.in.read();
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("发送消息出现异常...");
        } finally {
            // 7: 释放连接关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            if (connection != null && connection.isOpen()) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    };
    public static void main(String[] args) {
        // 启动三个线程去执行
        new Thread(runnable, "queue-1").start();
        new Thread(runnable, "queue-2").start();
        new Thread(runnable, "queue-3").start();
    }
}

8.6、RPC

7.7、Publisher Confirms

经出版商确认的可靠发布。

9、Springboot整合

引入依赖：

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

配置：

# 服务端口
server:
  port: 8080
# 配置rabbitmq服务
spring:
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    host: 47.104.141.27
    port: 5672

生产者

@Component
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // 1: 定义交换机
    private String exchangeName = "fanout_order_exchange";
    // 2: 路由key
    private String routeKey = "";
    public void makeOrder(Long userId, Long productId, int num) {
        // 1： 模拟用户下单
        String orderNumer = UUID.randomUUID().toString();
        // 2: 根据商品id productId 去查询商品的库存
        // int numstore = productSerivce.getProductNum(productId);
        // 3:判断库存是否充足
        // if(num >  numstore ){ return  "商品库存不足..."; }
        // 4: 下单逻辑
        // orderService.saveOrder(order);
        // 5: 下单成功要扣减库存
        // 6: 下单完成以后
        System.out.println("用户 " + userId + ",订单编号是：" + orderNumer);
        // 发送订单信息给RabbitMQ fanout
        rabbitTemplate.convertAndSend(exchangeName, routeKey, orderNumer);
    }
}

绑定关系：

public class DirectRabbitConfig {
    //队列 起名：TestDirectQueue
    @Bean
    public Queue emailQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);
        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("email.fanout.queue", true);
    }
    @Bean
    public Queue smsQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);
        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("sms.fanout.queue", true);
    }
    @Bean
    public Queue weixinQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);
        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("weixin.fanout.queue", true);
    }
    //Direct交换机 起名：TestDirectExchange
    @Bean
    public DirectExchange fanoutOrderExchange() {
        //  return new DirectExchange("TestDirectExchange",true,true);
        return new DirectExchange("fanout_order_exchange", true, false);
    }
    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    public Binding bindingDirect1() {
        return BindingBuilder.bind(weixinQueue()).to(fanoutOrderExchange()).with("");
    }
    @Bean
    public Binding bindingDirect2() {
        return BindingBuilder.bind(smsQueue()).to(fanoutOrderExchange()).with("");
    }
    @Bean
    public Binding bindingDirect3() {
        return BindingBuilder.bind(emailQueue()).to(fanoutOrderExchange()).with("");
    }
}

测试：

@SpringBootTest
class SpringbootRabbitmqFanoutProducerApplicationTests {
    @Autowired
    OrderService orderService;
    @Test
    public void contextLoads() throws Exception {
        for (int i = 0; i < 10; i++) {
            Thread.sleep(1000);
            Long userId = 100L + i;
            Long productId = 10001L + i;
            int num = 10;
            orderService.makeOrder(userId, productId, num);
        }
    }
}

消费者：

// bindings其实就是用来确定队列和交换机绑定关系
@RabbitListener(bindings =@QueueBinding(
        // email.fanout.queue 是队列名字，这个名字你可以自定随便定义。
        value = @Queue(value = "email.fanout.queue",autoDelete = "false"),
        // order.fanout 交换机的名字 必须和生产者保持一致
        exchange = @Exchange(value = "fanout_order_exchange",
                // 这里是确定的rabbitmq模式是：fanout 是以广播模式 、 发布订阅模式
                type = ExchangeTypes.FANOUT)
))
@Component
public class EmailService {
    // @RabbitHandler 代表此方法是一个消息接收的方法。该不要有返回值
    @RabbitHandler
    public void messagerevice(String message){
        // 此处省略发邮件的逻辑
        System.out.println("email-------------->" + message);
    }
}

10、过期TTL

过期时间TTL表示可以对消息设置预期的时间，在这个时间内都可以被消费者接收获取；过了之后消息将自动被删除。RabbitMQ可以对消息和队列设置TTL。目前有两种方法可以设置。

第一种方法是通过队列属性设置，队列中所有消息都有相同的过期时间。
第二种方法是对消息进行单独设置，每条消息TTL可以不同。

设置管道过期时间：

public class Producer {
    public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("47.104.141.27");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4: 从连接中获取通道channel
            channel = connection.createChannel();
            // 5: 申明队列queue存储消息
            /*
             *  如果队列不存在，则会创建
             *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
             *
             *  @params1： queue 队列的名称
             *  @params2： durable 队列是否持久化
             *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
             *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
             *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
             * */
            Map<String,Object> args2 = new HashMap<>();
            args2.put("x-message-ttl",5000);
            channel.queueDeclare("ttl.queue", true, false, false, args2);
            // 6： 准备发送消息的内容
            String message = "你好，学相伴！！！";
            Map<String, Object> headers = new HashMap<String, Object>();
            headers.put("x", "1");
            headers.put("y", "1");
            AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder()
                    .deliveryMode(2) // 传送方式
                    .priority(1)
                    .contentEncoding("UTF-8") // 编码方式
                    .expiration("3000") // 过期时间
                    .headers(headers).build(); //自定义属性
            // 7: 发送消息给中间件rabbitmq-server
            // @params1: 交换机exchange
            // @params2: 队列名称/routing
            // @params3: 属性配置
            // @params4: 发送消息的内容
            for (int i = 0; i <100 ; i++) {
                channel.basicPublish("", "ttl.queue", basicProperties, message.getBytes());
                System.out.println("消息发送成功!");
                Thread.sleep(1000);
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("发送消息出现异常...");
        } finally {
            // 7: 释放连接关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            if (connection != null && connection.isOpen()) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}

参数 x-message-ttl 的值必须是非负 32 位整数 (0 <= n <= 2^32-1) ，以毫秒为单位表示 TTL 的值。这样，值 6000 表示存在于队列中的当前消息将最多只存活 6 秒钟。

设置消息过期时间：

public class MessageTTLProducer {
    public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("47.104.141.27");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4: 从连接中获取通道channel
            channel = connection.createChannel();
            // 5: 申明队列queue存储消息
            /*
             *  如果队列不存在，则会创建
             *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
             *
             *  @params1： queue 队列的名称
             *  @params2： durable 队列是否持久化
             *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
             *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
             *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
             * */
            channel.queueDeclare("ttl.queue2", true, false, false, null);
            // 6： 准备发送消息的内容
            String message = "你好，学相伴！！！";
            Map<String, Object> headers = new HashMap<String, Object>();
            headers.put("x", "1");
            headers.put("y", "1");
            AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder()
                    .deliveryMode(2) // 传送方式
                    .priority(1)
                    .contentEncoding("UTF-8") // 编码方式
                    .expiration("5000") // 过期时间
                    .headers(headers).build(); //自定义属性
            // 7: 发送消息给中间件rabbitmq-server
            // @params1: 交换机exchange
            // @params2: 队列名称/routing
            // @params3: 属性配置
            // @params4: 发送消息的内容
            for (int i = 0; i <10 ; i++) {
                channel.basicPublish("", "ttl.queue2", basicProperties, message.getBytes());
                System.out.println("消息发送成功!");
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("发送消息出现异常...");
        } finally {
            // 7: 释放连接关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            if (connection != null && connection.isOpen()) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}

expiration 字段以微秒为单位表示 TTL 值。且与 x-message-ttl 具有相同的约束条件。因为 expiration 字段必须为字符串类型，broker 将只会接受以字符串形式表达的数字。
当同时指定了 queue 和 message 的 TTL 值，则两者中较小的那个才会起作用。

少用管道过期，尽量使用消息过期

11、消息确认机制

NONE值是禁用发布确认模式，是默认值
CORRELATED值是发布消息成功到交换器后会触发回调方法
SIMPLE值经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法，其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker;

# 配置rabbitmq服务
spring:
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    host: 47.104.141.27
    port: 5672
    publisher-confirm-type: correlated

12、死信队列

DLX，全称为Dead-Letter-Exchange , 可以称之为死信交换机，也有人称之为死信邮箱。当消息在一个队列中变成死信(dead message)之后，它能被重新发送到另一个交换机中，这个交换机就是DLX ，绑定DLX的队列就称之为死信队列。

消息变成死信，可能是由于以下的原因：

消息被拒绝
消息过期
队列达到最大长度

DLX也是一个正常的交换机，和一般的交换机没有区别，它能在任何的队列上被指定，实际上就是设置某一个队列的属性。当这个队列中存在死信时，Rabbitmq就会自动地将这个消息重新发布到设置的DLX上去，进而被路由到另一个队列，即死信队列。
要想使用死信队列，只需要在定义队列的时候设置队列参数 x-dead-letter-exchange 指定交换机即可。

13、集群

部署集群

官方文档：https://www.rabbitmq.com/clustering.html

主节点操作：

#停止应用
> sudo rabbitmqctl -n rabbit-1 stop_app
#目的是清除节点上的历史数据（如果不清除，无法将节点加入到集群）
> sudo rabbitmqctl -n rabbit-1 reset
#启动应用
> sudo rabbitmqctl -n rabbit-1 start_app

从节点操作：

# 停止应用
> sudo rabbitmqctl -n rabbit-2 stop_app
# 目的是清除节点上的历史数据（如果不清除，无法将节点加入到集群）
> sudo rabbitmqctl -n rabbit-2 reset
# 将rabbit2节点加入到rabbit1（主节点）集群当中【Server-node服务器的主机名】
> sudo rabbitmqctl -n rabbit-2 join_cluster rabbit-1@'Server-node'
# 启动应用
> sudo rabbitmqctl -n rabbit-2 start_app

验证节点状态

1	> sudo rabbitmqctl cluster_status -n rabbit-1

主节点，从节点账号共享

如果采用多机部署方式，需读取其中一个节点的cookie, 并复制到其他节点（节点之间通过cookie确定相互是否可通信）。cookie存放在/var/lib/rabbitmq/.erlang.cookie。
例如：主机名分别为rabbit-1、rabbit-2
1、逐个启动各节点
2、配置各节点的hosts文件( vim /etc/hosts)
ip1：rabbit-1
ip2：rabbit-2
其它步骤雷同单机部署方式

14、分布式事务

为不同的服务提供数据一致性。

方法：

两阶段提交（2PC）
补偿事务（TCC）
本地消息表（异步确保）
MQ事务消息

14.1、两阶段提交（2PC）

定义：两阶段提交（Two-Phase Commit，2PC）是一种分布式事务协议，用于确保在多个数据库或事务参与者之间的操作能够原子性地完成或回滚。这协议涉及两个主要阶段：准备阶段和提交（或回滚）阶段。

保证强一致性

基本工作流程：

准备阶段（Phase 1 - Prepare）：
- 协调者（Coordinator）向所有参与者发送准备请求，询问它们是否可以执行提交操作。
- 参与者收到请求后，执行本地事务的准备操作，但是并不提交事务。
- 参与者向协调者发送准备就绪或准备失败的响应。
提交阶段（Phase 2 - Commit）：
- 如果所有参与者都准备就绪，协调者向所有参与者发送提交请求。
- 参与者收到提交请求后，执行实际的事务提交操作。
- 如果有任何一个参与者在准备阶段失败，协调者会发送回滚请求给所有参与者，要求它们回滚之前的操作。

存在的问题

同步阻塞所有事务参与者在等待其它参与者响应的时候都处于同步阻塞状态，无法进行其它操作。
单点问题协调者在 2PC 中起到非常大的作用，发生故障将会造成很大影响。特别是在阶段二发生故障，所有参与者会一直等待状态，无法完成其它操作。
数据不一致在阶段二，如果协调者只发送了部分 Commit 消息，此时网络发生异常，那么只有部分参与者接收到 Commit 消息，也就是说只有部分参与者提交了事务，使得系统数据不一致。
太过保守任意一个节点失败就会导致整个事务失败，没有完善的容错机制。

14.2、补偿事务（TCC）

使用公司如：严选，阿里，蚂蚁金服。

TCC的核心思想是：针对每一个操作都需要注册一个和其相对应的确认和补偿的操作，他分为三个阶段Try、Confirm和Cancel

Try 阶段主要是对业务系统做检测及资源预留
Confirm 阶段主要是对业务系统做确认提交，Try阶段执行成功并开始执行Confirm阶段时，默认Confirm阶段是不会出错的。即：只要Try成功，Confirm一定成功。若Confirm阶段真的出错了，需引入重试机制或人工处理。
Cancel阶段主要是在业务执行错误，需要回滚的状态下执行的业务取消，预留资源释放。

优点：跟2PC比起来，实现以及流程相对简单了一些，但数据的一致性比2PC也要差一些
缺点：缺点还是比较明显的，在2,3步中都有可能失败。TCC属于应用层的一种补偿方式，所以需要程序员在实现的时候多写很多补偿的代码，在一些场景中，一些业务流程可能用TCC不太好定义及处理。

14.3、本地消息表（异步确保）

使用公司如：支付宝、微信支付主动查询支付状态，对账单的形式。

本地消息表与业务数据表处于同一个数据库中，这样就能利用本地事务来保证在对这两个表的操作满足事务特性，并且使用了消息队列来保证最终一致性。

在分布式事务操作的一方完成写业务数据的操作之后向本地消息表发送一个消息，本地事务能保证这个消息一定会被写入本地消息表中。
之后将本地消息表中的消息转发到 Kafka 等消息队列中，如果转发成功则将消息从本地消息表中删除，否则继续重新转发。
在分布式事务操作的另一方从消息队列中读取一个消息，并执行消息中的操作。

优点：一种非常经典的实现，避免了分布式事务，实现了最终一致性。
缺点：消息表会耦合到业务系统中，如果没有封装好的解决方案，会有很多杂活需要处理。

14.4、MQ事务消息

有一些第三方的MQ是支持事务消息的，比如RocketMQ，他们支持事务消息的方式也是类似于采用的二阶段提交，但是市面上一些主流的MQ都是不支持事务消息的，比如 Kafka 不支持。
以阿里的 RabbitMQ 中间件为例，其思路大致为：

第一阶段Prepared消息，会拿到消息的地址。第二阶段执行本地事务，第三阶段通过第一阶段拿到的地址去访问消息，并修改状态。
也就是说在业务方法内要想消息队列提交两次请求，一次发送消息和一次确认消息。如果确认消息发送失败了RabbitMQ会定期扫描消息集群中的事务消息，这时候发现了Prepared消息，它会向消息发送者确认，所以生产方需要实现一个check接口，RabbitMQ会根据发送端设置的策略来决定是回滚还是继续发送确认消息。这样就保证了消息发送与本地事务同时成功或同时失败。

优点：实现了最终一致性，不需要依赖本地数据库事务。
缺点：实现难度大，主流MQ不支持，RocketMQ事务消息部分代码也未开源。

14.5、RabbitMQ实现

问题：

分布式系统中，数据一致性是个难题，可以使用RabbitMQ进行事务管理。

解决方法：

使用rabbitmq

可靠生产：

创建一个消息冗余表，进行事务回滚。

可靠消费：

使用RabbitMQ的ACK机制，由消费者自身控制消息的重发、清楚和丢弃

基于MQ的分布式事务消息的死信队列消息转移 + 人工处理：

对丢弃的消息可以放入死信队列进行处理。

对于分布式事务问题：

尽量去避免分布式事务
尽量将非核心业务做成异步

例子

@Service
public class OrderService {
    @Autowired
    private OrderDataBaseService orderDataBaseService;
    // 创建订单
    @Transactional(rollbackFor = Exception.class) // 订单创建整个方法添加事务
    public void createOrder(Order orderInfo) throws Exception {
        // 1: 订单信息--插入丁订单系统，订单数据库事务
        orderDataBaseService.saveOrder(orderInfo);
        // 2：通過Http接口发送订单信息到运单系统
        String result = dispatchHttpApi(orderInfo.getOrderId());
        if(!"success".equals(result)) {
            throw new Exception("订单创建失败,原因是运单接口调用失败!");
        }
    }
    /**
     *  模拟http请求接口发送，运单系统，将订单号传过去 springcloud
     * @return
     */
    private String dispatchHttpApi(String orderId) {
        SimpleClientHttpRequestFactory factory  = new SimpleClientHttpRequestFactory();
        // 链接超时 > 3秒
        factory.setConnectTimeout(3000);
        // 处理超时 > 2秒
        factory.setReadTimeout(2000);
        // 发送http请求
        String url = "http://localhost:9000/dispatch/order?orderId="+orderId;
        RestTemplate restTemplate = new RestTemplate(factory);//异常
        String result = restTemplate.getForObject(url, String.class);
        return result;
    }
}

15、常见问题

15.1、幂等性问题

幂等性是指无论对一个操作执行多少次，结果都是一致的。在分布式系统中，由于网络延迟、故障恢复等原因，可能导致某个请求在系统中重复执行，因此保持操作的幂等性对于确保系统的一致性和正确性非常重要。

解决方法：

分布式锁：
- 使用分布式锁是一种防止重复执行的方法。在处理请求时，首先尝试获取分布式锁，如果成功则执行操作，如果失败则说明已经有其他请求在处理相同的操作，避免重复执行。
- 但是，分布式锁也可能引入性能开销和复杂性，并且需要注意死锁、锁超时等问题。
数据库唯一约束：
- 利用数据库的唯一约束是另一种处理幂等性问题的方法。在执行某个操作之前，首先检查数据库中是否已存在相同的记录，如果不存在则执行插入、更新等操作。
- 这种方法要求数据库中的数据设计支持唯一性，并且需要注意处理并发情况下的竞态条件。

选择使用哪种方法通常取决于具体的应用场景和需求：

分布式锁：
- 适用于需要在分布式环境中对操作进行全局性的控制，确保在整个系统中只有一个节点能够执行该操作。
数据库唯一约束：
- 适用于对于某个资源的唯一性要求，可以通过数据库的唯一性约束来保证，例如，在用户注册时，通过用户唯一的用户名或邮箱来确保用户的唯一性。

在实际应用中，有时也会结合使用这两种方法，根据具体场景来选择最适合的方案，以达到幂等性的目标。

15.2、RabbitMQ是基于channel而不是基于连接

为什么RabbitMQ是基于channel而不是基于连接？

资源消耗：
- 连接的创建和销毁通常比通道更昂贵。通过使用通道，可以在单个连接上创建多个通道，减少了连接的开销。
- 在大多数应用中，使用少量的连接和多个通道通常比使用多个连接更为高效。
并发性：
- 通道提供了更好的并发性。在单个连接上可以创建多个通道，而这些通道可以并行处理消息。
- 这对于在处理大量消息时提高性能和吞吐量非常重要，因为每个通道都可以独立工作，不受其他通道的影响。
管理和控制：
- 通过使用通道，可以更方便地控制和管理消息的流。每个通道可以独立地进行流量控制、错误处理和配置。
- 如果使用连接，所有通信都将在连接级别进行，这可能导致更复杂和难以管理的系统。
资源复用：
- 通过共享连接上的通道，可以更好地复用底层的TCP连接。这降低了与建立和断开TCP连接相关的开销，提高了整体效率。
轻量级：
- 通道的创建和销毁是轻量级的操作，这对于快速地适应变化的负载或连接断开再连接的情况更为灵活。

总体而言，基于通道的设计提供了更灵活、轻量级且高效的方式来处理消息。通过合理使用通道，RabbitMQ能够更好地适应各种负载和应用场景，提供更好的性能和可伸缩性。

15.3、queue队列到底在消费者创建还是生产者创建

queue队列到底在消费者创建还是生产者创建？

一般建议是在rabbitmq操作面板创建。这是一种稳妥的做法。
按照常理来说，确实应该消费者这边创建是最好，消息的消费是在这边。这样你承受一个后果，可能我生产在生产消息可能会丢失消息。
在生产者创建队列也是可以，这样稳妥的方法，消息是不会出现丢失。
如果你生产者和消费都创建的队列，谁先启动谁先创建，后面启动就覆盖前面的。

16、参考

笔记网址：https://www.kuangstudy.com/zl/rabbitmq

官方文档：https://www.rabbitmq.com/documentation.html

1、中间件

1.1、常用中间件

1.2、基于消息中间件的分布式系统的架构

2、消息队列协议

2.1、协议

2.2、常见消息队列协议

3、消息队列对比

4、高可用与高可靠

4.1、高可用

4.1.1、集群模式

4.2、高可靠

5、安装

5.1、Linux安装

5.2、Docker安装

5.3、RabbitMQ常用命令

6、角色分类

7、核心组成

8、七种工作模式

8.1、simple

8.2、work

8.3、Publish/Subscribe

8.4、Routing

8.5、Topics

8.6、RPC

7.7、Publisher Confirms

9、Springboot整合

10、过期TTL

11、消息确认机制

12、死信队列

13、集群

14、分布式事务

14.1、两阶段提交（2PC）

14.2、补偿事务（TCC）

14.3、本地消息表（异步确保）

14.4、MQ事务消息

14.5、RabbitMQ实现

15、常见问题

15.1、幂等性问题

15.2、RabbitMQ是基于channel而不是基于连接

15.3、queue队列到底在消费者创建还是生产者创建

16、参考