在线问答5

这个屌丝很懒，什么也没留下！

热门标签

RabbitMQ笔记

作者：在线问答5 | 2024-07-30 11:52:46

踩

RabbitMQ笔记

文章目录

基本概念
RabbitMQ

基本概念

一、中间件

1.1什么是中间件

中间件（Middleware）是处于操作系统和应用程序之间的软件，也有人认为它应该属于操作系统中的一部分。人们在使用中间件时，往往是一组中间件集成在一起，构成一个平台（包括开发平台和运行平台），但在这组中间件中必须要有一个通信中间件，即中间件=平台+通信，这个定义也限定了只有用于分布式系统中才能称为中间件，同时还可以把它与支撑软件和实用软件区分开来。

1.2单体架构

在企业开发的中，大部分的初期架构都采用的是单体架构的模式进行架构，而这种架构的典型的特点：就是把所有的业务和模块，源代码，静态资源文件等都放在一个一工程中，如果其中的一个模块升级或者迭代发生一个很小变动都会重新编译和重新部署项目。这种的架构存在的问题就是：

.存在问题

1：耦合度太高
2：运维的成本过高
3：不易维护
4：服务器的成本高
5：升级架构的复杂度也会增大

1.3微服务架构

就是一个请求由服务器端的多个服务（服务或者系统）协同处理完成

存在问题
1：学习成本高，技术栈过多
2：运维成本和服务器成本增高
3：人员的成本也会增高
4：项目的负载度也会上升
5：面临的错误和容错性也会成倍增加
6：占用的服务器端口和通讯的选择的成本高
7：安全性的考虑和因素逼迫可能选择RMI/MQ相关的服务器端通讯。

好处
1：服务系统的独立，占用的服务器资源减少和占用的硬件成本减少，
确切的说是：可以合理的分配服务资源，不造成服务器资源的浪费
2：系统的独立维护和部署，耦合度降低，可插拔性。
3：系统的架构和技术栈的选择可以变的灵活（而不是单纯的选择java）
4：弹性的部署，不会造成平台因部署造成的瘫痪和停服的状态。

二、消息队列协议

2.1什么是协议

所谓协议是指：
1：计算机底层操作系统和应用程序通讯时共同遵守的一组约定，只有遵循共同的约定和规范，系统和底层操作系统之间才能相互交流。
2：和一般的网络应用程序的不同它主要负责数据的接受和传递，所以性能比较的高。
3：协议对数据格式和计算机之间交换数据都必须严格遵守规范。

协议是在tcp/ip协议基础之上构建的一种约定成俗的规范和机制、它的主要目的可以让客户端（应用程序 java，go）进行沟通和通讯。并且这种协议下规范必须具有持久性，高可用，高可靠的性能。

消息中间件采用的并不是http协议，而常见的消息中间件协议有：OpenWire、AMQP、MQTT、Kafka，OpenMessage协议。

2.2为什么消息中间件不直接使用http协议

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

2.3网络协议三要素

1.语法。语法是用户数据与控制信息的结构与格式,以及数据出现的顺序。
2.语义。语义是解释控制信息每个部分的意义。它规定了需要发出何种控制信息,以及完成的动作与做出什么样的响应。
3.时序。时序是对事件发生顺序的详细说明。

2.4不同协议之间对比

AMQP

AMQP：(全称：Advanced Message Queuing Protocol) 是高级消息队列协议。由摩根大通集团联合其他公司共同设计。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等。
特性：
1：分布式事务支持。
2：消息的持久化支持。
3：高性能和高可靠的消息处理优势。

MQTT

MQTT协议：（Message Queueing Telemetry Transport）消息队列是IBM开放的一个即时通讯协议，物联网系统架构中的重要组成部分。
特点：
1：轻量
2：结构简单
3：传输快，不支持事务
4：没有持久化设计。
应用场景：
1：适用于计算能力有限
2：低带宽
3：网络不稳定的场景。

OpenMessage协议

是近几年由阿里、雅虎和滴滴出行、Stremalio等公司共同参与创立的分布式消息中间件、流处理等领域的应用开发标准。
特点：
1：结构简单
2：解析速度快
3：支持事务和持久化设计。

Kafka

Kafka协议是基于TCP/IP的二进制协议。消息内部是通过长度来分割，由一些基本数据类型组成。
特点是：
1：结构简单
2：解析速度快
3：无事务支持
4：有持久化设计

三、消息队列特性

3.1不同消息队列的持久化

在这里插入图片描述

3.2消息分发策略

在这里插入图片描述

3.3消息队列的高可用

所谓高可用是指：是指系统可以无故障低持续运行，比如一个系统突然崩溃，报错，异常等等并不影响线上业务的正常运行，出错的几率极低，就称之为：高可靠。
在高并发的业务场景中，如果不能保证系统的高可靠，那造成的隐患和损失是非常严重的。
如何保证中间件消息的可靠性呢？可以从两个方面考虑：
1：消息的传输：通过协议来保证系统间数据解析的正确性。
2：消息的存储可靠：通过持久化来保证消息的可靠性。

RabbitMQ

一、基础

1.1是什么

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

1.2命令

常用命令

启动服务

systemctl start rabbitmq-server

重启服务

systemctl restart rabbitmq-server

查看服务状态

systemctl status rabbitmq-server

停止服务

systemctl stop rabbitmq-server

开机启动服务

systemctl enable rabbitmq-server

安装web控制台插件

rabbitmq-plugins enable rabbitmq_management

用户相关

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 ".” “." ".”

linux排查命令

more xxx.log 查看日记信息
netstat -naop | grep 5672 查看端口是否被占用
ps -ef | grep 5672 查看进程
systemctl stop 服务

docker相关命令

docker博客

https://blog.csdn.net/weixin_44860598/article/details/105163585

创建并运行容器

默认用户名密码guest

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

方式二
启动容器时命名为myrabbit，同时映射rabbitmq和宿主机的端口，并设置用户名和密码为admin

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

启动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

1.3角色分类

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

二、核心组成部分

在这里插入图片描述

2.1核心概念

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,消息队列，保存消息并将它们转发给消费者。

2.2运行流程

三、支持消息的模式

3.1简单模式

流程

生产者生产数据到队列->消费者消费

注意点

如果没有指定交换机，会存在一个默认的交换机
是否要持久化durable 。所谓持久化消息是否存盘，如果false 、会存盘，但是会随从重启服务会丢失。

生产者

public class Producer {


    public static void main(String[] args) {

        // 所有的中间件技术都是基于tcp/ip协议基础之上构建新型的协议规范，只不过rabbitmq遵循的是amqp
        // ip port

        // 1: 创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.1.106");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");

        //springboot ---rabbitmq

        Connection connection = null;
        Channel channel = null;
        try {
            // 2: 创建连接Connection Rabbitmq为什么是基于channel去处理而不是链接? 长连接----信道channel
            connection = connectionFactory.newConnection("生成者");
            // 3: 通过连接获取通道Channel
            channel = connection.createChannel();
            // 4: 通过通创建交换机，声明队列，绑定关系，路由key,发送消息，和接收消息
            String queueName = "queue1";

            /*
             * @params1 队列的名称
             * @params2 是否要持久化durable=false 所谓持久化消息是否存盘，如果false 非持久化 true是持久化? 非持久化会存盘吗？ 会存盘，但是会随从重启服务会丢失。
             * @params3 排他性，是否是独占独立
             * @params4 是否自动删除，随着最后一个消费者消息完毕消息以后是否把队列自动删除
             * @params5 携带附属参数
             */
            channel.queueDeclare(queueName, true, false, false, null);
            // 5: 准备消息内容
            String message = "Hello zyp!!!";
            // 6: 发送消息给队列queue
            // @params1: 交换机  @params2 队列、路由key @params 消息的状态控制  @params4 消息主题
            // 面试题：可以存在没有交换机的队列吗？不可能，虽然没有指定交换机但是一定会存在一个默认的交换机。
            channel.basicPublish("", queueName, null, message.getBytes());

            System.out.println("消息发送成功!!!");
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            // 7: 关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            // 8: 关闭连接

            if (connection != null && connection.isOpen()) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }


    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

消费者

public class Consumer {


    public static void main(String[] args) {

        // 所有的中间件技术都是基于tcp/ip协议基础之上构建新型的协议规范，只不过rabbitmq遵循的是amqp
        // ip port

        // 1: 创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.1.106");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");

        Connection connection = null;
        Channel channel = null;
        try {
            // 2: 创建连接Connection
            connection = connectionFactory.newConnection("消费者");
            // 3: 通过连接获取通道Channel
            channel = connection.createChannel();
            // 4: 通过通创建交换机，声明队列，绑定关系，路由key,发送消息，和接收消息


            // true = ack 正常的逻辑是没问题 死循环 rabbit 重发策略
            // false = nack 消息这在消费消息的时候可能会异常和故障
            final  Channel channel2 = channel;
            channel2.basicConsume("queue1", false, new DeliverCallback() {
                public void handle(String consumerTag, Delivery message) throws IOException {
                    try {
                        System.out.println("收到消息是" + new String(message.getBody(), "UTF-8"));
                        channel2.basicAck(message.getEnvelope().getDeliveryTag(),false);
                    }catch (Exception ex){
                        ex.printStackTrace();
                        // 三次确认 -- reject + sixin
                    }

                }
            }, new CancelCallback() {
                public void handle(String consumerTag) throws IOException {
                    System.out.println("接受失败了...");
                }
            });

            System.out.println("开始接受消息");



        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            // 7: 关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            // 8: 关闭连接

            if (connection != null && connection.isOpen()) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }


    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

3.2fanout模式

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

3.3Direct模式

图解

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

每个exchange和queue之间可以通过路由key来进行绑定，便于分类筛选。

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ntBZ2Qeq-1626098492688)(C:\Users\鹏鹏\AppData\Roaming\Typora\typora-user-images\image-20210709165645339.png)]$

生产者核心代码

// 3: 从连接工厂中获取连接
connection = connectionFactory.newConnection("生产者");
// 4: 从连接中获取通道channel
channel = connection.createChannel();
// 6： 准备发送消息的内容
String message = " 你好，zyp";
// 交换机
String  exchangeName = "amq.direct";
// 交换机的类型 direct/topic/fanout/headers
String exchangeType = "direct";

// 如果你用界面把queueu 和 exchange的关系先绑定话，你代码就不需要在编写这些声明代码可以让代码变得更加简洁，但是不容读懂
// 如果用代码的方式去声明，我们要学习一下
// 7： 声明交换机 所谓的持久化就是指，交换机会不会随着服务器重启造成丢失，如果是true代表不丢失，false重启就会丢失
channel.exchangeDeclare(exchangeName,exchangeType,true);

// 8： 声明队列
channel.queueDeclare("queue5",true,false,false,null);
channel.queueDeclare("queue6",true,false,false,null);
channel.queueDeclare("queue7",true,false,false,null);

// 9：绑定队列和交换机的关系
channel.queueBind("queue5",exchangeName,"order");
channel.queueBind("queue6",exchangeName,"order");
channel.queueBind("queue7",exchangeName,"course");

channel.basicPublish(exchangeName, "order", null, message.getBytes());
System.out.println("消息发送成功!");
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

消费者核心代码

// 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();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

3.4Topic模式

图解

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

在这里插入图片描述

生产者核心代码

// 3: 从连接工厂中获取连接
connection = connectionFactory.newConnection("生产者");
// 4: 从连接中获取通道channel
channel = connection.createChannel();

// 5： 准备发送消息的内容
String message = "hello zyp";

// 6：准备交换机
String exchangeName = "amq.topic";
// 7: 定义路由key
String routeKey = "c.a.b";
// 8: 指定交换机的类型
String type = "topic";
// 7: 发送消息给中间件rabbitmq-server
// @params1: 交换机exchange
// @params2: 队列名称/routingkey
// @params3: 属性配置
// @params4: 发送消息的内容
//  #.course.* queue3
// *.order.# queue2 ta
// com.order.course.xxx collecion
channel.basicPublish(exchangeName, routeKey, null, message.getBytes());
System.out.println("消息发送成功!");
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

消费者核心代码

// 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() {
    public void handle(String s, Delivery delivery) throws IOException {
        System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));
    }
}, new CancelCallback() {
    public void handle(String s) throws IOException {
    }
});
System.out.println(queueName + "：开始接受消息");
System.in.read();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

3.5Headers模式

特点：Headers模式是Topic模式上的一种叠加，增加了选择参数的模式
queue里需要指定参数
路由器通过key和Headers条件筛选queue

在这里插入图片描述

3.6Work模式 - 轮询模式（Round-Robin）

特点：一个消费者一条，按均分配

生产者核心代码

// 3: 从连接工厂中获取连接
connection = connectionFactory.newConnection("生产者");
// 4: 从连接中获取通道channel
channel = connection.createChannel();
// 6： 准备发送消息的内容
//===============================end topic模式==================================
for (int i = 1; i <= 20; i++) {
    //消息的内容
    String msg = "zyp:" + i;
    // 7: 发送消息给中间件rabbitmq-server
    // @params1: 交换机exchange
    // @params2: 队列名称/routingkey
    // @params3: 属性配置
    // @params4: 发送消息的内容
    channel.basicPublish("", "queue1", null, msg.getBytesjava());
}
System.out.println("消息发送成功!");
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

消费者1核心代码

// 6： 定义接受消息的回调
Channel finalChannel = channel;
//finalChannel.basicQos(1);

finalChannel.basicConsume("queue1", true, new DeliverCallback() {
    @Override
    public void handle(String s, Delivery delivery) throws IOException {
        try{
            System.out.println("Work1-收到消息是：" + new String(delivery.getBody(), "UTF-8"));
            Thread.sleep(1000);
        }catch(Exception ex){
            ex.printStackTrace();
        }
    }
}, new CancelCallback() {
    @Override
    public void handle(String s) throws IOException {
    }
});
System.out.println("Work1-开始接受消息");
System.in.read();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

消费者2核心代码

// 3: 从连接工厂中获取连接
connection = connectionFactory.newConnection("消费者-Work2");
// 4: 从连接中获取通道channel
channel = connection.createChannel();
// 5: 申明队列queue存储消息
/*
 *  如果队列不存在，则会创建
 *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
 *
 *  @params1： queue 队列的名称
 *  @params2： durable 队列是否持久化
 *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
 *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
 *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
 * */
// 这里如果queue已经被创建过一次了，可以不需要定义
//channel.queueDeclare("queue1", false, true, false, null);
// 同一时刻，服务器只会推送一条消息给消费者
//channel.basicQos(1);
// 6： 定义接受消息的回调
Channel finalChannel = channel;
//finalChannel.basicQos(1);
finalChannel.basicConsume("queue1", true, new DeliverCallback() {
    @Override
    public void handle(String s, Delivery delivery) throws IOException {
        try{
            System.out.println("Work2-收到消息是：" + new String(delivery.getBody(), "UTF-8"));
            Thread.sleep(200);
        }catch(Exception ex){
            ex.printStackTrace();
        }
    }
}, new CancelCallback() {
    @Override
    public void handle(String s) throws IOException {
    }
});
System.out.println("Work2-开始接受消息");
System.in.read();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

结果：消费者1和消费者2接收的数量一样

3.7Work模式 - 公平分发（Fair Dispatch）

根据消费者的消费能力进行公平分发，处理快的处理的多，处理慢的处理的少；按劳分配；

生产者核心代码

// 3: 从连接工厂中获取连接
connection = connectionFactory.newConnection("生产者");
// 4: 从连接中获取通道channel
channel = connection.createChannel();
// 6： 准备发送消息的内容
//===============================end topic模式==================================
for (int i = 1; i <= 20; i++) {
    //消息的内容
    String msg = "zyp:" + i;
    // 7: 发送消息给中间件rabbitmq-server
    // @params1: 交换机exchange
    // @params2: 队列名称/routingkey
    // @params3: 属性配置
    // @params4: 发送消息的内容
    channel.basicPublish("", "queue1", null, msg.getBytes());
}
System.out.println("消息发送成功!");
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

消费者1核心代码

 // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("消费者-Work1");
            // 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.basicQos(1);
            finalChannel.basicConsume("queue1", false, new DeliverCallback() {
                public void handle(String s, Delivery delivery) throws IOException {
                    try{
                        System.out.println("Work1-收到消息是：" + new String(delivery.getBody(), "UTF-8"));
                        Thread.sleep(1000);
                        // 改成手动应答
                        finalChannel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
                    }catch(Exception ex){
                        ex.printStackTrace();
                    }
                }
            }, new CancelCallback() {
                public void handle(String s) throws IOException {
                }
            });
            System.out.println("Work1-开始接受消息");
            System.in.read();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

消费者2核心代码

// 3: 从连接工厂中获取连接
connection = connectionFactory.newConnection("消费者-Work2");
// 4: 从连接中获取通道channel
channel = connection.createChannel();
// 5: 申明队列queue存储消息
/*
 *  如果队列不存在，则会创建
 *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
 *
 *  @params1： queue 队列的名称
 *  @params2： durable 队列是否持久化
 *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
 *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
 *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
 * */
// 这里如果queue已经被创建过一次了，可以不需要定义
//channel.queueDeclare("queue1", false, true, false, null);
// 同一时刻，服务器只会推送一条消息给消费者
//channel.basicQos(1);
// 6： 定义接受消息的回调
Channel finalChannel = channel;
finalChannel.basicQos(1);
finalChannel.basicConsume("queue1", true, new DeliverCallback() {
    @Override
    public void handle(String s, Delivery delivery) throws IOException {
        try{
            System.out.println("Work2-收到消息是：" + new String(delivery.getBody(), "UTF-8"));
            Thread.sleep(500);
            // 一定使用我们的手动应答
            //finalChannel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        }catch(Exception ex){
            ex.printStackTrace();
        }
    }
}, new CancelCallback() {
    @Override
    public void handle(String s) throws IOException {
    }
});
System.out.println("Work2-开始接受消息");
System.in.read();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

结果：消费者1明显比消费者2接收的多

rabbitmq

四、RabbitMQ的使用场景

异步执行
高内聚，低耦合
流量的削峰
分布式事务的可靠消费和可靠生产
索引、缓存、静态化处理的数据同步
流量监控
日志监控（ELK）
下单、订单分发、抢票

五、过期时间TTL

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

第一种是把消息队列设置成过期类型的队列，web界面会有TTL标识。设置完过期时间后，此队列的所有消息到期后都会过期。如果绑定了死信队列，则会把过期的消息移到死信队列里面。
第二种是给单个消息设置过期。
如果上述两种方法同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。

TTL标识

六、死信队列

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

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

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

七、内存磁盘的监控

7.1内存警告

当内存的使用超过了配置的阈值或磁盘剩余空间低于配置的阈值时，RabbitMQ会暂时阻塞客户端的连接，并且停止接收从客户端发来的消息，以此来避免服务器的崩溃，客户端与服务端的心态检测机制也会失效。

如果出现了问题，需要赶紧配置加内存或磁盘大小，同时检查代码是否有循环等操作。

7.2当内存使用超过了配置的阈值

解决：可以通过命令和配置文件的方式调整内存大小

命令方式

rabbitmqctl set_vm_memory_high_watermark
rabbitmqctl set_vm_memory_high_watermark absolute 50MB

farction为内存阈值。默认情况是:0.4/2GB。范围建议调整为本机内存大小的0.4-0.7之间。超过后就会触发内存警告。通过命令的方式重启后会失效。

配置方式

当前配置文件：/etc/rabbitmq/rabbitmq.conf

使用relative相对值进行设置fraction,建议取值在04~0.7之间，不建议超过0.7
vm_memory_high_watermark.relative = 0.6
使用absolute的绝对值的方式，但是是KB,MB,GB,对应的命令如下
vm_memory_high_watermark.absolute = 2GB

7.3当磁盘剩余空间低于阈值

默认情况下：磁盘预警为50MB的时候会进行预警。表示当前磁盘空间第50MB的时候会阻塞生产者并且停止内存消息换页到磁盘的过程。
这个阈值可以减小，但是不能完全的消除因磁盘耗尽而导致崩溃的可能性。比如在两次磁盘空间的检查空隙内，第一次检查是：60MB ，第二检查可能就是1MB,就会出现警告。

命令方式

rabbitmqctl set_disk_free_limit <disk_limit>
rabbitmqctl set_disk_free_limit memory_limit
disk_limit：固定单位 KB MB GB
fraction ：是相对阈值，建议范围在:1.0~2.0之间。（相对于内存）

配置方式

disk_free_limit.relative = 3.0
disk_free_limit.absolute = 50mb

7.4内存换页

在某个节点及内存阻塞生产之前，会将队列中的消息从内存转移到磁盘，以释放内存空间。持久化和非持久化的消息都会写入磁盘中，其中持久化的消息本身就在磁盘中有一个副本，所以在转移的过程中持久化的消息会先从内存中清除掉。

默认情况下，内存到达的阈值是50%时就会换页处理。
也就是说，在默认情况下该内存的阈值是0.4的情况下，当内存超过0.4*0.5=0.2时，会进行换页动作。

比如有1000MB内存，当内存的使用率达到了400MB,已经达到了极限，但是因为配置的换页内存0.5，这个时候会在达到极限400mb之前，会把内存中的200MB进行转移到磁盘中。从而达到稳健的运行。

可以通过设置 vm_memory_high_watermark_paging_ratio 来进行调整

vm_memory_high_watermark.relative = 0.4
vm_memory_high_watermark_paging_ratio = 0.7（设置小于1的值）

八、用RabbitMQ实现分布式事务

场景：用户下单后，需要经过订单服务和派送服务。而这两个服务之间是隔离的，用着不同的JVM。要保证这次下单业务的执行，需要使用分布式事务。

在这里插入图片描述

8.1系统调用之间的回滚问题

如果处理订单服务和数据库服务在一个事务中。处理完订单服务，处理数据库服务时，需要调用远程的数据库服务，此时如果数据库服务成功，但远程调用时由于网络等原因，调用失败，出现事务回滚。那么会出现数据库服务执行而订单服务回滚后没有执行，会产生数据不一致问题。

8.2消息的可靠生产问题

生产者产生订单时，同时会在一个冗余表生成数据。然后把数据发送到消息队列中。使用RabbitMQ提供的publisher,Confirm应答机制，收到ack后修改冗余数据中的状态。如果没有收到ack，则用其他方式把消息存储起来，并定时重发。
在这里插入图片描述

核心代码

订单业务

@Service
public class MQOrderService {
   @Autowired
   private OrderDataBaseService orderDataBaseService;
   @Autowired
   private OrderMQService orderMQService;
   // 创建订单
   public void createOrder(Order orderInfo) throws Exception {
      // 1: 订单信息--生成订单和冗余订单信息
      orderDataBaseService.saveOrder(orderInfo);
      // 2：用rabbitmq推送消息
      orderMQService.sendMessage(orderInfo);
   }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

保存订单具体逻辑

@Service
@Transactional(rollbackFor = Exception.class)
public class OrderDataBaseService {
	@Autowired
	private JdbcTemplate jdbcTemplate;
	/**
	 * 生成订单信息
	 */
	public void saveOrder(Order order) throws Exception{
		// 定义保存sql
		String sqlString = "insert into ksd_order(order_id,user_id,order_content)values(?,?,?)";
		// 1：添加订单记录
		int count = jdbcTemplate.update(sqlString,order.getOrderId(),order.getUserId(),order.getOrderContent());
		if(count!=1) {
			throw new Exception("订单创建失败，原因[数据库操作失败]");
		}
		//因为在下单可能会会rabbit会出现宕机，就引发消息是没有放入MQ.为来消息可靠生产，对消息做一次冗余
		saveLocalMessage(order);
	}
	
	/**
	 * 保存冗余订单信息
	 * @param order
	 */
	public  void saveLocalMessage(Order order) throws Exception{
		// 定义保存sql
		String sqlString = "insert into ksd_order_message(order_id,order_content,status,unique_id)values(?,?,?,?)";
		// 添加运动记录
		int count = jdbcTemplate.update(sqlString,order.getOrderId(),order.getOrderContent(),0,1);
		if(count!=1) {
			throw new Exception("出现异常，原因[数据库操作失败]");
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

ack应答和rabbitmq推送消息

@Service
public class OrderMQService {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private JdbcTemplate jdbcTemplate;

    //@PostConstruct注解好多人以为是Spring提供的。其实是Java自己的注解。
    //Java中该注解的说明：@PostConstruct该注解被用来修饰一个非静态的void（）方法。被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，
    // 并且只会被服务器执行一次。PostConstruct在构造函数之后执行，init（）方法之前执行。
    @PostConstruct
    public void regCallback() {
        // 消息发送成功以后，给予生产者的消息回执,来确保生产者的可靠性
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("cause:"+cause);
                // 如果ack为true代表消息已经收到
                String orderId = correlationData.getId();

                if (!ack) {
                    // 这里可能要进行其他的方式进行存储，并定时重发。
                    System.out.println("MQ队列应答失败，orderId是:" + orderId);
                    return;
                }

                try {
                    String updatesql = "update ksd_order_message set status = 1 where order_id = ?";
                    int count = jdbcTemplate.update(updatesql, orderId);
                    if (count == 1) {
                        System.out.println("本地消息状态修改成功，消息成功投递到消息队列中...");
                    }
                } catch (Exception ex) {
                    System.out.println("本地消息状态修改失败，出现异常：" + ex.getMessage());
                }
            }
        });
    }


    public void sendMessage(Order order) {
        // 通过MQ发送消息
        rabbitTemplate.convertAndSend("order_fanout_exchange", "", JsonUtil.obj2String(order),
                new CorrelationData(order.getOrderId()));
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

定时重发

@EnableScheduling
public class TaskService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Scheduled(cron = "0 0 0/2 ?")
    public void sendMessage(){
        // 把消息为0的状态消息重新查询出来，投递到MQ中。
        List<Order> orderList = orderService.selectOrderMessage(0);
        for (Order order : orderList) {
            rabbitTemplate.convertAndSend("order-fanout_exchange","",order);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

8.3消息的可靠消费问题

消费者监听的队列得到消息后，执行派送业务。如果业务失败，采用MQ提供的重试机制或者try，catch+basicNack(tag,false,false)+死信队列的方式。第一次失败，catch捕获后，把消息放入死信队列。监听死信队列的业务中重新执行派送业务，此时需要考虑幂等性问题。如果再执行失败，可以考虑发短信预警，人工干预。同时把消息转移到别的存储DB

在这里插入图片描述

核心代码
派送服务

@Service
@Transactional(rollbackFor = Exception.class)
public class DispatchService {
    @Autowired
    private JdbcTemplate jdbcTemplate;
    public void dispatch(String orderId) throws Exception {
        // 定义保存sql
        String sqlString = "insert into ksd_dispather_order(order_id,dispatch_id,status,order_content,user_id)values(?,?,?,?,?)";
        // 添加运动记录
        int count = jdbcTemplate.update(sqlString, orderId, UUID.randomUUID().toString(), 0, "木子鱼买了一个泡面", "1");

        if (count != 1) {
            throw new Exception("订单创建失败，原因[数据库操作失败]");
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

消费者监听

@Service
public class OrderMqConsumer {
    @Autowired
    private DispatchService dispatchService;
    private int count = 1;
    
    // 解决消息重试的集中方案：
    // 1: 控制重发的次数 + 死信队列
    // 2: try+catch+手动ack
    // 3: try+catch+手动ack + 死信队列处理 + 人工干预
    @RabbitListener(queues = {"order.queue"})
    public void messageconsumer(String ordermsg, Channel channel,
                                CorrelationData correlationData,
                                @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        try {
            // 1:获取消息队列的消息
            System.out.println("收到MQ的消息是: " + ordermsg + ",count = " + count++);
            //  2: 获取订单服务的信息
            Order order = JsonUtil.string2Obj(ordermsg, Order.class);
            // 3: 获取订单id
            String orderId = order.getOrderId();
            // 4：保存运单
            dispatchService.dispatch(orderId);
            // 3：手动ack告诉mq消息已经正常消费
            System.out.println(1 / 0); //出现异常

            channel.basicAck(tag, false);
        } catch (Exception ex) {
            //如果出现异常的情况下,根据实际的情况去进行重发
            //重发一次后，丢失，还是日记，存库根据自己的业务场景去决定
            //参数1：消息的tag  参数2：false 多条处理 参数3：requeue 重发
            // false 不会重发，会把消息打入到死信队列
            // true 的会会死循环的重发，建议如果使用true的话，不加try/catch否则就会造成死循环
            channel.basicNack(tag, false, false);// 死信队列
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

死信队列监听

@Service
public class DeadMqConsumer {
    @Autowired
    private DispatchService dispatchService;
    // 解决消息重试的集中方案：
    // 1: 控制重发的次数 + 死信队列
    // 2: try+catch+手动ack
    // 3: try+catch+手动ack + 死信队列处理 + 人工干预
    @RabbitListener(queues = {"dead.order.queue"})
    public void messageconsumer(String ordermsg, Channel channel,
                                CorrelationData correlationData,
                                @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        try {
            // 1:获取消息队列的消息
            System.out.println("收到MQ的消息是: " + ordermsg );
            //  2: 获取订单服务的信息
            Order order = JsonUtil.string2Obj(ordermsg, Order.class);
            // 3: 获取订单id
            String orderId = order.getOrderId();
            // 幂等性问题
            //int count = countOrderById(orderId);
            // 4：保存运单
            //if(count==0)dispatchService.dispatch(orderId);
            //if(count>0)dispatchService.updateDispatch(orderId);
             dispatchService.dispatch(orderId);
            // 3：手动ack告诉mq消息已经正常消费
            channel.basicAck(tag, false);
        } catch (Exception ex) {
            System.out.println("人工干预");
            System.out.println("发短信预警");
            System.out.println("同时把消息转移别的存储DB");
            channel.basicNack(tag, false,false);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

8.4优缺点以及建议

优点

1、通用性强
2、拓展方便
3、耦合度低，方案也比较成熟

缺点
1、基于消息中间件，只适合异步场景
2、消息会延迟处理，需要业务上能够容忍

建议

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

本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/在线问答5/article/detail/903642