2021-04-03 狂神说Java之 rabbitmq学习

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了2021-04-03 狂神说Java之 rabbitmq学习，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含63088字，纯文字阅读大概需要91分钟。

内容图文

中间件及单体架构

消息中间件

在实际的项目中，大部分的企业项目开发中，在早期都采用的是单体的架构模式

单体架构

把所有的业务和模块、源代码、静态资源文件等都放在一个工程中，如果其中的一个模块升级或迭代发生一个很小的变动都会重新编译和重新部署项目。（耦合度高）

这种的架构存在的问题就是：

• 耦合度太高
• 运维的成本过高
• 不易维护
• 服务器的成本高（服务器资源的浪费）
• 以及升级架构的复杂度也会增大（局限性）

这样就有后续的分布式架构系统

分布式架构

单体架构一个请求由一个系统共同完成，分布式架构系统就是一个请求由服务器的多个服务（服务或者系统）协同处理完成（跨JVM的请求调用）

和单体架构不同的是，单体架构师一个请求发起jvm调度线程（tomcat线程池）分配线程thread 来处理请求指导释放

分布式系统是：一个请求是由多个系统共同协同完成，jvm和环境都可能是独立

比如：单体架构就是建造一个小房子很快就能搞定，但要建造大型的建筑就必须是各个环节的协同和分布，这样的目的也是项目发展到后期的时候需要部署和思考的问题

分布式系统存在的问题：

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

好处：

• 服务系统的独立，占用的服务器资源减少和占用的硬件成本减少（可以合理的分配服务资源，不造成服务器资源的浪费）
• 系统的独立维护和部署，耦合度降度，可插拔性
• 系统的架构和技术栈的选择可以变得灵活
• 弹性的部署，不会造成平台因部署造成的瘫痪和停服的状态

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

消息中间件（是否具有通讯的能力，是否拥有高性能，易灵活，支持多平台，持久化功能，高可靠性，容错性，消息的分发策略等）

• 利用可靠的消息传递机制进行系统和系统直接的通讯
• 通过提供消息传递和消息的排队机制，它可以在分布式系统环境下扩展进程间的通讯

消息中间件的应用场景

1、跨系统的数据传递

2、高并发的流量削峰

3、数据的分发和异步处理

方法栈调用的过程：串行执行

并行执行：（异步编程）

4、大数据分布与传递

5、分布式事务

常见的消息中间件

ActiveMQ，RabbitMQ，Kafka，RocketMQ等

消息中间件的本质及设计

本质：它是一种接受数据、接受请求、存储数据、发送数据等功能的技术服务

MQ消息队列：负责数据的传接受，存储和传递，所以性能要高于普通的服务和技术，要遵循协议

消息中间件的核心组成部分

1、消息的协议

2、消息持久化机制

3、消息的分发策略

4、消息的高可用、高可靠

5、消息的容错机制

消息队列的协议

什么是协议

消息中间件负责数据的传递、存储、和分发消息三个部分，数据的存储和分发的过程中要遵循某种约定成俗的规范，无论是采用底层的TCP/IP，UDP协议还是其他在这之上自己去开发的协议等，这些约定成俗的规范就称之为：协议

协议

1、计算机底层操作系统和应用程序通讯时共同遵守的一组约定，只有遵循共同的约定和规范，系统和底层操作系统之间才能相互交流（为了数据的流向、通讯、接受、分发）

2、和一般的网络应用程序的不同，它主要负责数据的接受和传递，所以性能比较的高

3、协议对数据格式和计算机之间交换数据都必须严格遵循规范

网络协议的三要素

1、语法：语法是用户数据与控制信息的结构与格式以及数据出现的顺序

2、语义：语义是解释控制信息每个部分的意义。他规定了需要发出何种控制信息，以及完成的动作与做出什么样的响应

3、时序：时序是对事件发生顺序的详细说明

Http请求协议：

1、语法：http规定了请求报文和响应报文的格式

2、语义：客户端主动发起请求称之为请求（这是一种定义，同时你发起的是post/get请求）

3、时序：一个请求对应一个响应（一定先有请求再有响应，这个是时序）

消息中间件采用的协议有：openwire，AMQP，MQTT，Kafka，OpenMessage协议

面试：为什么消息中间件不直接使用http协议？

1、因为http请求报文头和响应报文头是比较复杂的，包含了cookie，数据的加密解密，状态码，响应码等附加的功能，但是对于一个消息而言，不需要这么复杂，只要负责数据传输、存储、分发就行，一定要准球的是高性能。尽量简洁、快速

2、大部分情况下http大部分都是短链接（在实际的交互过程中，一个请求到响应很有可能会中断，中断以后就不会进行持久化，会造成请求的丢失）。这样就不利于消息中间件的业务场景，因为消息中间件可能是一个长期的获取消息的过程，出现问题和故障要对数据或消息进行持久化等，目的是为了保证消息和数据的高可靠和稳健的运行。（长链接：当服务器出现故障，重启后依然可以进行数据的传递）

AMQP协议

Advance Message Queuing Protocol 是高级消息队列协议，提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准。为面向消息的中间件设计（基于此协议的客户端与消息中间件可传递消息，并不受客户端、中间件不同产品，不同的开发语言等条件的限制）。Erlang中的实现有rabbitMQ等

特性：

1、分布式事务支持

2、消息的持久化支持

3、高性能和高可靠的消息处理优势

支持者: rabbitMQ. activeMQ

MQTT协议

Message Queueing Telementry Transport 消息队列是IBM 开发的一个即时通讯协议，物联网系统架构中的重要组成部分

特点：

1、轻量

2、结构简单

3、传输快，不支持事务

4、没有持久化设计

应用场景：

1、适用于计算能力有限

2、低带宽

3、网络不稳低的场景

支持者：RabbitMQ activeMQ

OpenMessage协议

阿里，雅虎，滴滴出行等公司共同参与创立的分布式消息中间件、流处理等领域的应用开发标准。

特点：

1、结构简单

2、解析速度快

3、支持事务和持久化设计

Kafka 协议

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

特点：

1、结构简单

2、解析速度快

3、无事务支持

4、有持久化设计

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

消息队列的持久化

持久化

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

消息分发策略

MQ消息队列有如下的角色：

1、生产者

2、存储消息

3、消费者

轮询分发（公平的，不会因为服务器的快慢、性能的高低而造成数据的倾斜）

公平分发（能者多劳，会造成数据的倾斜）

消息队列的高可用和高可靠

什么是高可用机制

高可用：是指产品在规定的条件和规定的时刻或事件内处于可执行规定功能状态的能力

当业务量增加时，请求也过大，一台消息中间件服务器会触及硬件（cpu\内存\磁盘）的极限，一台消息服务器以及无法满足业务的请求，所以消息中间件必须支持集群部署，来达到高可用的目的

集群模式

1 master-slave 主从共享数据的部署方式

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

2 -Master -slave 主从同步部署方式

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

3 - 多主集群同步部署模式

多写多读的机制，和上面区别不是很大，但他的写入可以往任意节点去写入

4 -多主集群转发部署模式

（元数据的转发机制，元数据不会把消息的本体放入元数据信息里面，只会把队列名、链接信息等存入其中）如果你插入的数据是broker-1 中，元数据信息会存储数据的相关描述和记录存放的位置（队列）。它会对描述信息（元数据）信息进行同步，如果消费者在broker-2 中进行消费，发现没有对应的消息，可以从其他对应的元数据信息中去查询，然后返回对应的消息信息。（减少空间存储和资源浪费）

（例如：买火车票或者黄牛票，第一个黄牛有顾客说要买的演唱会门票，但是没有，但他回去联系其他的黄牛询问，如果有就返回）

5 - Master-slave与Breoker-cluster组合的方案

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

集群模式最终目的都是为来保证：消息服务器不会挂掉，出现故障依然可以抱着消息服务继续使用

要么消息共享、要么消息同步、要么元数据共享

什么是高可靠机制

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

如何保证中间件消息的可靠性

1、消息的传输：通过协议来保证系统间数据解析的正确性

2、消息的存储可靠：通过持久化来保证消息的可靠性

RabbitMQ 入门和安装

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

mac

安装Brew

/usr/bin/ruby -e "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/master/install)"

下载安装erlang和RabbitMQ

  // 安装erlang环境
	brew install erlang
	// 更新brew资源
  brew update
  // 执行安装
  brew install rabbitmq

出现版本号表示erlang安装完成

erl -v

安装RabbitMq的可视化监控插件

// 切换到MQ目录,注意你的安装版本可能不是3.8.14
cd /usr/local/Cellar/rabbitmq/3.8.14/
// 启用rabbitmq management插件
sudo sbin/rabbitmq-plugins enable rabbitmq_management

配置环境变量

sudo vi /etc/profile
//加入以下两行
export RABBIT_HOME=/usr/local/Cellar/rabbitmq/3.8.14
export PATH=$PATH:$RABBIT_HOME/sbin

:wq! 强制保存退出
// 立即生效
source /etc/profile

后台启动rabbitMQ

 // 后台启动
 sudo rabbitmq-server -detached  
  // 查看状态
 sudo rabbitmqctl status 
  // 访问可视化监控插件的界面
  // 浏览器内输入 http://localhost:15672,默认的用户名密码都是guest,登录后可以在Admin那一列菜单内添加自己的用户
  sudo rabbitmqctl stop 关闭

文档查看：

• mac安装brew(亲测)：
  https://blog.csdn.net/yuanshangshenghuo/article/details/106599836
  Mac安装erlang和rabbitmq:
  https://www.jianshu.com/p/59b97f388268
  mac + RabbitMQ 安装：https://www.jianshu.com/p/60c358235705

所遇问题：

• brew install opencv@2 时报错 Error: Can’t create update lock in /usr/local/var/homebrew/locks!
  https://blog.csdn.net/u012135425/article/details/89095183

• Mac安装Homebrew的正确姿势:
  https://www.jianshu.com/p/e0471aa6672d?utm_campaign=hugo

• chown: /usr/local: Operation not permitted：

• https://blog.csdn.net/qq_15722433/article/details/99735837

RabbitMQ 管理界面及权限操作

// 安装rabbitmq management插件
sudo sbin/rabbitmq-plugins enable rabbitmq_management
// 或是
rabbitmq-plugins enable rabbitmq_management

浏览器内输入 http://localhost:15672,默认的用户名密码都是guest,(只能在localhost本机下访问)，需要添加远程登录的用户

安装完成以后，重启服务即可

// 后台启动
 sudo rabbitmq-server -detached  
// Linux 启动
systemctl restart rabbitmq-server

注意：在对应服务器（阿里云，腾讯云等）的安全组中开放15672的端口！！！

// 新增用户 用户名 密码
sudo rabbitmqctl add_user admin admin
// 设置用户admin分配操作权限
sudo rabbitmqctl set_user_tags admin administrator

用户级别：

• 1、administrator：（超级管理员）可以登录控制台、查看所有信息、可以对rabbitmq进行管理
• 2、monitoring ：监控者。登录控制台，查看所有信息
• 3、policymaker：策略制定者 登录控制台，指定策略
• 4、management: 普通管理员 登录控制台
• 5、none ： 不能访问management plugin

// 为用户添加资源权限（授予admin 访问虚拟机所有根节点的权限）
sudo rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

小结

# 添加账号密码
rabbitmqctl add_user 账号 密码
# 赋予角色
rabbitmqctl set_user_tags 账号 administrator
# 修改密码
rabbitmqctl change_password 用户名 新密码
# 删除用户
rabbitmqctl delete_user 用户名
# 查看用户清单
rabbitmqctl list_users
# 为用户设置 administrator角色
rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"
rabbitmqctl set_permissions -p / root ".*" ".*" ".*"
# 启动服务器 
rabbitmq-server -detached
# 关闭服务器
rabbitmqctl stop
# 查看所有队列信息
rabbitmqctl list_queues
# 关闭应用
rabbitmqctl stop_app
# 启动应用，和上述关闭命令配合使用，达到清空队列的目的
 rabbitmqctl start_app
# 清除所有队列
rabbitmqctl reset
# 服务器重启
systemctl restart rabbitmq-server

• rabbitmq 管理及常用命令 https://blog.csdn.net/wochunyang/article/details/52449559

Docker安装RabbitMQ

虚拟化容器技术-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/docer- 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/doker/daemon.json <<-'EOF'
{
	"registry-mirrors":["https://0wrdwnn6.mirror.aliyuncs.com"]
}
EOF
sudo systemctl daemon-reload
sudo systemctl restart docker

RabbitMQ的角色分类

none ：

不能访问management plugin(无法登录界面)

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

列出自己可以通过AMQP登入的虚拟机

查看自己的虚拟机节点 virtual hosts 的queues，exchange和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

查看创建permissions

关闭所有用户的connections

RabbitMQ入门案例

simple简单模式

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

生产者—>队列—>消费者

构建一个maven工程

导入依赖

 <!--   java 原生rabbitmq依赖-->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.10.0</version>
</dependency>

生产者

// 简单模式----生产者
public class Producer {
    public static void main(String[] args) {
        // 所有的中间件技术都是基于tcp/ip协议基础上构建新型的协议规范,rabbitmq遵循的是amqp
        // 协议遵循 ip port

        // 1、创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 通过连接工厂设置账号密码
        connectionFactory.setHost("127.0.0.1");
        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，发送消息和接受消息
            String queueName = "queue1";
            /**声明队列
             * @params1 queue 队列名，
             * @params2 durable 是否需要持久化，durable false (消息是否存盘）
             *          如果false 非持久化 true是持久化吗？非持久化会存盘吗？
             * @params3 exclusive 是否要具有排他性，是否是独占队列
             * @params4 autoDelete是否要自动删除，随着最后一个消费者消费完毕之后是否把队列自动删除
             * @params5 arguments 携带的附属参数(设置队列的有效期，消息最大长度）
             */
            channel.queueDeclare(queueName, false, false, false, null);
            // 5、准备消息内容
            String message = "hello";
             // 6、发送消息给队列 queue (交换机 ,队列路由key[队列名字]，消息的状态控制【持久化】，消息主题【消息内容】）
            channel.basicPublish("", queueName, null, message.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 7、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 8、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

消费者

// 简单模式----消费者
public class Customer {
    public static void main(String[] args) {
        // 所有的中间件技术都是基于tcp/ip协议基础上构建新型的协议规范,rabbitmq遵循的是amqp
        // 协议遵循 ip port

        // 1、创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 通过连接工厂设置账号密码
        connectionFactory.setHost("127.0.0.1");
        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，发送消息和接受消息
            channel.basicConsume("queue1", true, new DeliverCallback() {
                public void handle(String consumerTag, Delivery message) throws IOException {
                    System.out.println("收到消息是：" + new String(message.getBody(), "UTF-8"));
                }
            }, new CancelCallback() {
                public void handle(String consumerTag) throws IOException {
                    System.out.println("接受消息失败了。。。。");
                }
            });
            System.out.println("开始接受消息");
            // 进行阻断，接受消息不关闭
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 7、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 8、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

非持久化消息队列，随着服务器的关闭，会被移除，而持久化消息队列不会被移除

持久化的消息会存盘，非持久化的消息也会存盘，但会随着服务器重启而丢失

AMQP

Advance Message Queuing Protocol (高级消息队列协议) 是应用层协议的一个开发标准，为面向消息的中间件设计

AMQP生产者流转过程

面试：rabbitmq 为什么是基于channel去处理的而不是连接？

长连接-----信道channel

AMQP消费者流转过程

broker (节点)

RabbitMQ的核心组成部分

• Server：Broker ，接受客户端的连接，实现AMQP实体服务，安装rabbitmq-server

• Connection：连接，应用程序与Broker的网络连接 TCP/IP 三次握手和四次挥手（会出现性能损耗，长连接）

• Channel：网络信道，几乎所有的操作都在channel中进行，channel是进行信息读写的通道，客户端可以建立对各channel，每个channel代表一个会话任务

• Message：消息，服务与应用程序之间传送的数据，由Properties和body组成，properties是对消息进行修饰，（如：消息的优先级、延迟等高级特性）body则是消息体的内容

• Virtual Host：虚拟地址，由于进行逻辑隔离，最上层的消息路由，一个虚拟主机路由可以由若干个交换机和队列，同一个虚拟主机里面不能有相同名字的交换机（做隔离，虚拟机节点）

• exchange：交换机，接受消息，根据路由键发送消息到绑定的队列（不具备消息存储的能力）

• Bindings：绑定，交换机和队列之间的虚拟连接，binding中可以保护多个routing key

• Routing key:路由key，是一个路由规则，虚拟机可以用它来确定如何路由一个特定消息（条件，类似于数据库的条件where routekey= course）

• queue：队列，也称为message queue 消息队列，保存消息并将它转发给消费者

面试题：可以存在没有交换机的队列吗？exchange

不可能，虽然没有指定交换机，但是一定会存在一个默认的交换机

RabbitMQ运行流程

RabbitMQ 支持消息的模式

简单模式 Simple

理解

模拟生产者往交换机里面发送消息

模拟消费者接受消息

注意：

1、无论是简单模式还是工作队列模式，都会有一个默认交换机存在

2、发送消息是交换机来发送，不是队列发送，交换机会接收到消息然后推送给队列，消费者会自动监听和订阅队列，把消息进行推送

3、如果队列没有声明交换机，一定绑定的默认交换机

4、生产过程中进行消息预览时 一定要选择：Nack message requeue true

（ack 会当成真实的消费进行移除，造成数据的丢失！！）

发布与订阅Fanout模式-Publish/Subscribe

理解（web界面模拟）

建立一个交换机

创建队列

绑定关系

• 队列界面绑定
  

• 交换机界面绑定（都可）
  
  在交换机里发送消息
  
  
  接受消息
  

发布订阅模式具体实现

Fanout --是一种广播机制，它是没有路由key的模式

生产者

// 生产者
// 如果在界面已经把交换机和队列的关系绑定好，程序中就可以不需要绑定了
public class Producer {
    public static void main(String[] args) {

        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        // 通过连接工厂设置账号密码
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、通过连接获取通道channel
            channel = connection.createChannel();
            // 5、准备发送消息的内容
            String message = "hello fanout_exchange java";
            // 6、准备交换机
            String exchange = "fanout-exchange";
            // 7、定义路由key
            String routeKey = "";
            // 8、指定交换机的类型
            String type = "fanout";
            // 9、发送消息给中间件rabbitmq-server
            // （交换机exchange，队列名称/routingkey,属性配置，发送消息的内容）
            channel.basicPublish(exchange, routeKey, null, message.getBytes());
            System.out.println("消息发送成功！！！");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 10、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 11、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

消费者

// 消费者
public class Customer {
    private static Runnable runnable = new Runnable() {
        public void run() {
            // 1、创建连接工程
            ConnectionFactory connectionFactory = new ConnectionFactory();
            // 2、设置连接属性
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            // 虚拟访问节点
            connectionFactory.setVirtualHost("/");
            // 获取消息队列的名字
            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 3、从连接工厂中获取连接
                connection = connectionFactory.newConnection("生产者");
                // 4、从连接中获取通道channel
                channel = connection.createChannel();
                // 5、申明队列queue 存储信息
                // 如果队列不存在，则会创建  rabbintmq不允许创建两个相同的队列名称，否则会报错
                // 这里如果queue 已经被创建过一次，可以不需要定义
                // channel.queueDeclare("queue1",false,false,false,null);
                // 6、定义接受消息的回调
                Channel finalChannel = channel;
                finalChannel.basicConsume(queueName, true, new DeliverCallback() {
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        System.out.println(message.getEnvelope().getDeliveryTag());
                        System.out.println("收到消息是：" + new String(message.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("接受消息失败了。。。。");
                    }
                });
                System.out.println(queueName+"开始接受消息");
                // 进行阻断，接受消息不关闭
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 7、关闭通道（先关通道，再关连接）
                if(channel!=null && channel.isOpen()){
                    try {
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                // 8、关闭连接
                if(connection!=null && connection.isOpen()){
                    try {
                        connection.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };
    public static void main(String[] args) {
        // 启动三个线程去执行
        new Thread(runnable,"queue1").start();
        new Thread(runnable,"queue2").start();
        new Thread(runnable,"queue3").start();
    }
}

打印

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
queue1开始接受消息
queue2开始接受消息
queue3开始接受消息
1
1
收到消息是：hello fanout_exchange java
1
收到消息是：hello fanout_exchange java
收到消息是：hello fanout_exchange java

路由Routing（Direct）模式

理解

默认交换机模式就是direct模式

创建交换机

绑定队列，并创建路由key


发送消息

Direct模式具体实现

生产者

// 生产者
// 如果在界面已经把交换机和队列的关系绑定好，程序中就可以不需要绑定了
public class Producer {
    public static void main(String[] args) {

        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        // 通过连接工厂设置账号密码
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、通过连接获取通道channel
            channel = connection.createChannel();
            // 5、准备发送消息的内容
            String message = "hello fanout_exchange java";
            // 6、准备交换机
            String exchange = "direct_exchange";
            // 7、定义路由key
            String routeKey = "email";
            // 8、指定交换机的类型
            String type = "direct";
            // 9、发送消息给中间件rabbitmq-server
            // （交换机exchange，队列名称/routingkey,属性配置，发送消息的内容）
            channel.basicPublish(exchange, routeKey, null, message.getBytes());
            System.out.println("消息发送成功！！！");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 10、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 11、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

消费者

// 消费者
public class Customer {
    private static Runnable runnable = new Runnable() {
        public void run() {
            // 1、创建连接工程
            ConnectionFactory connectionFactory = new ConnectionFactory();
            // 2、设置连接属性
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            // 虚拟访问节点
            connectionFactory.setVirtualHost("/");
            // 获取消息队列的名字
            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 3、从连接工厂中获取连接
                connection = connectionFactory.newConnection("生产者");
                // 4、从连接中获取通道channel
                channel = connection.createChannel();
                // 5、申明队列queue 存储信息
                // 如果队列不存在，则会创建  rabbintmq不允许创建两个相同的队列名称，否则会报错
                // 这里如果queue 已经被创建过一次，可以不需要定义
                // channel.queueDeclare("queue1",false,false,false,null);
                // 6、定义接受消息的回调
                Channel finalChannel = channel;
                finalChannel.basicConsume(queueName, true, new DeliverCallback() {
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        System.out.println(queueName+"收到消息是：" + new String(message.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("接受消息失败了。。。。");
                    }
                });
                System.out.println(queueName+"开始接受消息");
                // 进行阻断，接受消息不关闭
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 7、关闭通道（先关通道，再关连接）
                if(channel!=null && channel.isOpen()){
                    try {
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                // 8、关闭连接
                if(connection!=null && connection.isOpen()){
                    try {
                        connection.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };
    public static void main(String[] args) {
        // 启动三个线程去执行
        new Thread(runnable,"queue1").start();
        new Thread(runnable,"queue2").start();
        new Thread(runnable,"queue3").start();
    }
}

Topic（主题）模式–Topics

理解

可以支持模糊匹配的路由key

创建交换机

绑定队列 （# 表示0个或多个或多级 *有且只有一级）

发送消息

Queue1 接收到消息（以com开头后面是com.xxx/com.xx.xx/ com都可以）

• com.course.order — queue1 queue2 queue3 收到
• com.course.order.xxx.xxx — queue1 queue3 收到
• course — 都收不到
• com.course.order.user.test — queue1 queue3 queue4 收到

Topic模式具体实现

生产者

public class Producer {
    public static void main(String[] args) {

        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        // 通过连接工厂设置账号密码
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、通过连接获取通道channel
            channel = connection.createChannel();
            // 5、准备发送消息的内容
            String message = "hello fanout_exchange java";
            // 6、准备交换机
            String exchange = "topic_exchange";
            // 7、定义路由key
            String routeKey = "com.order.test.xxx";
            // 8、指定交换机的类型
            String type = "topic";
            // 9、发送消息给中间件rabbitmq-server
            // （交换机exchange，队列名称/routingkey,属性配置，发送消息的内容）
            channel.basicPublish(exchange, routeKey, null, message.getBytes());
            System.out.println("消息发送成功！！！");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 10、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 11、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

消费者等同

参数Headers模式

理解

代码中传入参数的位置

创建交换机

绑定队列

发送

完整的声明创建方式

生产者

// 生产者
public class Producer {
    public static void main(String[] args) {

        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        // 通过连接工厂设置账号密码
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、通过连接获取通道channel
            channel = connection.createChannel();
            // 5、准备发送消息的内容
            String message = "hello 自己创建交换机 绑定关系 队列";
            // 6、准备交换机
            String exchangeName = "direct_message_exchange";
            // 指定交换机的类型  direct/topic/fanout/headers
            String exchaneType = "direct";
            // 声明交换机，所谓的持久化 durable（交换机会不会随着服务器重启造成丢失，true就不会丢失，false代表会丢失）
            channel.exchangeDeclare(exchangeName,exchaneType,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");
            // 10、发送消息给中间件rabbitmq-server
            // （交换机exchange，队列名称/routingkey,属性配置，发送消息的内容）
            channel.basicPublish(exchangeName, "order", null, message.getBytes());
            System.out.println("消息发送成功！！！");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 10、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 11、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

消费者

// 消费者
public class Customer {
    private static Runnable runnable = new Runnable() {
        public void run() {
            // 1、创建连接工程
            ConnectionFactory connectionFactory = new ConnectionFactory();
            // 2、设置连接属性
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            // 虚拟访问节点
            connectionFactory.setVirtualHost("/");
            // 获取消息队列的名字
            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 3、从连接工厂中获取连接
                connection = connectionFactory.newConnection("生产者");
                // 4、从连接中获取通道channel
                channel = connection.createChannel();
                // 5、申明队列queue 存储信息
                // 如果队列不存在，则会创建  rabbintmq不允许创建两个相同的队列名称，否则会报错
                // 这里如果queue 已经被创建过一次，可以不需要定义
                // channel.queueDeclare("queue1",false,false,false,null);
                // 6、定义接受消息的回调
                Channel finalChannel = channel;
                finalChannel.basicConsume(queueName, true, new DeliverCallback() {
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        System.out.println(queueName+"收到消息是：" + new String(message.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("接受消息失败了。。。。");
                    }
                });
                System.out.println(queueName+"开始接受消息");
                // 进行阻断，接受消息不关闭
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 7、关闭通道（先关通道，再关连接）
                if(channel!=null && channel.isOpen()){
                    try {
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                // 8、关闭连接
                if(connection!=null && connection.isOpen()){
                    try {
                        connection.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };
    public static void main(String[] args) {
        // 启动三个线程去执行
        new Thread(runnable,"queue5").start();
        new Thread(runnable,"queue6").start();
        new Thread(runnable,"queue7").start();
    }
}

在开发中可以通过界面进行绑定，声明，在Java代码中就可以移除

在消费者和生产过程中，如果交换机不存在则会出现异常

队列未进行注册也会出现异常

声明/绑定在消费者声明或生产者声明都是可以的

work模式

当有多个消费者时，我们的消息会被哪个消费者消费呢，该如何均衡消费者消费信息的多少呢？（主要有两种模式）

• 1、轮询模式的分发：一个消费者一条，按均分配
• 2、公平分发：根据消费者的消费能力进行公平分发，处理快的处理的多，处理慢的处理的少，按劳分配；

轮询模式（Round-Robin）

消息的分配模式时一个消费者分配一条，直至消息消费完成

不会因为服务器资源处理消息速度的快慢，而产生消费不对等性

生产者

// 生产者
public class Producer {
    public static void main(String[] args) {

        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        // 通过连接工厂设置账号密码
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、通过连接获取通道channel
            channel = connection.createChannel();
            // 5、准备发送消息的内容
            for (int i = 0; i < 20; i++) {
                String msg = "hello"+i;
                // 发送消息给中间件rabbitmq-server
                // （交换机exchange，队列名称/routingkey,属性配置，发送消息的内容）
                channel.basicPublish("", "queue1", null, msg.getBytes());
                Thread.sleep(1000);
            }
            System.out.println("消息发送成功！！！");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 10、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 11、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

消费者

// 消费者
public class Work1 {
    public static void main(String[] args) {
        // 1、创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、从连接中获取通道channel
            channel = connection.createChannel();
            // 5、申明队列queue 存储信息
            // 如果队列不存在，则会创建  rabbintmq不允许创建两个相同的队列名称，否则会报错
            // 这里如果queue 已经被创建过一次，可以不需要定义
            // channel.queueDeclare("queue1",false,false,false,null);
            // 6、定义接受消息的回调
            Channel finalChannel = channel;
//                finalChannel.basicQos(1);
            finalChannel.basicConsume("queue1", true, new DeliverCallback() {
                public void handle(String consumerTag, Delivery message) throws IOException {
                    try {
                        System.out.println("work1收到消息是：" + new String(message.getBody(), "UTF-8"));
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, new CancelCallback() {
                public void handle(String consumerTag) throws IOException {
                    System.out.println("接受消息失败了。。。。");
                }
            });
            System.out.println("work1开始接受消息");
            // 进行阻断，接受消息不关闭
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 7、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 8、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

public class Work2 {
    public static void main(String[] args) {
        // 1、创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、从连接中获取通道channel
            channel = connection.createChannel();
            // 5、申明队列queue 存储信息
            // 如果队列不存在，则会创建  rabbintmq不允许创建两个相同的队列名称，否则会报错
            // 这里如果queue 已经被创建过一次，可以不需要定义
            // channel.queueDeclare("queue1",false,false,false,null);
            // 6、定义接受消息的回调
            Channel finalChannel = channel;
//                finalChannel.basicQos(1);
            finalChannel.basicConsume("queue1", true, new DeliverCallback() {
                public void handle(String consumerTag, Delivery message) throws IOException {
                    try {
                        System.out.println("work2收到消息是：" + new String(message.getBody(), "UTF-8"));
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, new CancelCallback() {
                public void handle(String consumerTag) throws IOException {
                    System.out.println("接受消息失败了。。。。");
                }
            });
            System.out.println("work2开始接受消息");
            // 进行阻断，接受消息不关闭
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 7、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 8、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

公平分发

生产者

// 生产者
public class Producer {
    public static void main(String[] args) {

        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        // 通过连接工厂设置账号密码
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、通过连接获取通道channel
            channel = connection.createChannel();
            // 5、准备发送消息的内容
            for (int i = 0; i < 20; i++) {
                String msg = "hello"+i;
                // 发送消息给中间件rabbitmq-server
                // （交换机exchange，队列名称/routingkey,属性配置，发送消息的内容）
                channel.basicPublish("", "queue1", null, msg.getBytes());
            }
            System.out.println("消息发送成功！！！");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 10、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 11、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

消费者

公平分发：一定要将应答机制改为手动应答
定义指标： qos = 1(默认情况下是没有设置的，表示每次从队列重去除的数据量，不应设置太大值需要根据内存、cpu的占用率）

public class Work1 {
    public static void main(String[] args) {
        // 1、创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、从连接中获取通道channel
            channel = connection.createChannel();
            // 5、申明队列queue 存储信息
            // 如果队列不存在，则会创建  rabbintmq不允许创建两个相同的队列名称，否则会报错
            // 这里如果queue 已经被创建过一次，可以不需要定义
            // channel.queueDeclare("queue1",false,false,false,null);
            // 6、定义接受消息的回调
            final Channel finalChannel = channel;
            // 指标定义出来 qos = 1(默认情况下是没有设置的，表示每次从队列重去除的数据量）
            finalChannel.basicQos(1);
            // 公平分发------一定要将应答机制改为手动应答， autoAck ：false
            finalChannel.basicConsume("queue1", false, new DeliverCallback() {
                public void handle(String consumerTag, Delivery message) throws IOException {
                    try {
                        System.out.println("work1收到消息是：" + new String(message.getBody(), "UTF-8"));
                        Thread.sleep(1000);
                        // 改为手动应答--单条消费
                        finalChannel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, new CancelCallback() {
                public void handle(String consumerTag) throws IOException {
                    System.out.println("接受消息失败了。。。。");
                }
            });
            System.out.println("work1开始接受消息");
            // 进行阻断，接受消息不关闭
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 7、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 8、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

public class Work2 {
    public static void main(String[] args) {
        // 1、创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        // 虚拟访问节点
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3、从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4、从连接中获取通道channel
            channel = connection.createChannel();
            // 5、申明队列queue 存储信息
            // 如果队列不存在，则会创建  rabbintmq不允许创建两个相同的队列名称，否则会报错
            // 这里如果queue 已经被创建过一次，可以不需要定义
            // channel.queueDeclare("queue1",false,false,false,null);
            // 6、定义接受消息的回调
            final Channel finalChannel = channel;
            finalChannel.basicQos(1);
            // 将应答机制改为手动应答， autoAck ：false
            finalChannel.basicConsume("queue1", false, new DeliverCallback() {
                public void handle(String consumerTag, Delivery message) throws IOException {
                    try {
                        System.out.println("work2收到消息是：" + new String(message.getBody(), "UTF-8"));
                        Thread.sleep(200);
                        // 改为手动应答--单条消费
                        finalChannel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, new CancelCallback() {
                public void handle(String consumerTag) throws IOException {
                    System.out.println("接受消息失败了。。。。");
                }
            });
            System.out.println("work2开始接受消息");
            // 进行阻断，接受消息不关闭
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 7、关闭通道（先关通道，再关连接）
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 8、关闭连接
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

RabbitMQ使用场景

为什么使用rabbitMQ
分布式架构使用，模块与模块之间要进行互通和协同，然后就考虑使用什么样的消息队列，选择了rabbitMQ，是异步的，多线程，是一个分发机制，多线程机制，可以让我们的网站性能成倍的提升，因为异步就可以让处理数据的能力变得高效和稳健。在开发的过程中把各个服务进行分裂，就可以解耦

01解耦、削峰、异步

同步异步的问题（串行）

串行方式：将订单信息写入数据库成功后，再发送邮件、短信。以上三个任务全部完成后，再返回客户端

代码

从上至下的代码方式（穿行执行：所有方法的时间总和）

public void makeOrder{
	// 1 保存订单
	orderService.saveOrder();
	// 2 发送短信服务
	messageService.sendSMS("order");// 1-2s
	// 3 发送Email服务
	emailService .sendEmail("order");//1-2s
	// 4 发送app服务
	appService.sendApp("order");
}

并行方式 异步线程池

并行方式：将订单信息写入数据库成功后，发送注册邮件的同时，发送短信。以上三个任务全部完成后，返回客户端。（并行的方式可以提高处理时间）

代码

public void makeOrder{
	// 1 保存订单
	orderService.saveOrder();
	// 2 相关发送
	relationMessage();
}
public void relationMessage(){
	// 异步
	theadpool submit(new Callable<Object>(){

            public Object call() throws Exception {
               messageService.sendSMS("order");
            }
        });

	theadpool submit(new Callable<Object>(){

            public Object call() throws Exception {
               emailService .sendEmail("order");
            }
        });

	theadpool submit(new Callable<Object>(){

            public Object call() throws Exception {
               appService.sendApp("order"）;
            }
        });
}

存在问题：
1、耦合度高
2、需要自己写线程池维护成本高
3、出现了消息可能会丢失：需要自己做消息补偿
4、如何保证消息的可靠性：需要自己写
5、如果服务器承载不了：需要自己写高可用

异步消息队列方式

下单当作生产者，服务变成消费者

好处：
1、完全解耦，用MQ建立桥接
2、有独立的线程池和运行模型
3、出现了消息可能会丢失：MQ有持久化功能
４、如何保证消息的可靠性：死信队列和消息转移等
５、如果服务器承载不了：需要自己写高可用，HA镜像模型高可以

02高内聚,低耦合

解耦:

03流量的削峰

RabbitMQ与springboot整合

Faout模式(配置类方式绑定关系)

生产者 (创建一个生产者工程)

<!--rabbitMQ 的start依赖-->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>

配置文件

# 服务端口
server:
  port: 8080
# 配置rabbitmq服务 本机服务可以不用配置
spring:
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    host: 127.0.0.1
    port: 5672

代码

@Service
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 模拟用户下单
     * @param userId
     * @param productId
     * @param num
     */
    public void makeOrder(String userId,String productId,int num){
        // 1:根据商品id查询库存是否充足
        // 2:保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功:"+orderId);
        // 3:通过MQ来完成消息的分发
        // 参数1:交换机  参数2:路由Key/queue队列名称 参数3:消息内容
        String exchangName = "fanout_order_exchange";
        String rouetingKey = "";
        rabbitTemplate.convertAndSend(exchangName,rouetingKey,orderId);
    }
}

config

@Configuration
public class RabbitMQConfiguration {

    // 1:声明注册fanout 模式的交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        // 参数名 是否持久化 是否自动删除
        return new FanoutExchange("fanout_order_exchange",true,false);
    }
    // 2:声明队列 sms.fanout.queue duanxin.fanout.queue email.fanout.queue
    @Bean
    public Queue smsQueue(){
        return new Queue("sms.fanout.queue",true);
    }

    @Bean
    public Queue duanxinQueue(){
        return new Queue("duanxin.fanout.queue",true);
    }

    @Bean
    public Queue emailQueue(){
        return new Queue("email.fanout.queue",true);
    }
    // 3:完成绑定关系 队列和交换机绑定
    @Bean
    public Binding smsBingding(){
        return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
    }

    @Bean
    public Binding duanxinBingding(){
        return BindingBuilder.bind(duanxinQueue()).to(fanoutExchange());
    }

    @Bean
    public Binding emailBingding(){
        return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
    }
}

测试

@SpringBootTest
class SpringbootOrderRabbitmqProducerApplicationTests {

	@Autowired
	private OrderService orderService;
	@Test
	void contextLoads() {
		orderService.makeOrder("1","1",12);
	}

}

消费者 (创建一个消费者工程)

# 服务端口
server:
  port: 8081
# 配置rabbitmq服务 本机服务可以不用配置
spring:
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    host: 127.0.0.1
    port: 5672

代码

@RabbitListener(queues = {"email.fanout.queue"})
@Component
public class FanoutEmailConsumer {

    @RabbitHandler
    public void reviseMessage(String message) {
        System.out.println("email fanout ----接收到了订单信息是:-->" + message);
    }
}

@RabbitListener(queues = {"sms.fanout.queue"})
@Component
public class FanoutSMSConsumer {

    @RabbitHandler
    public void reviseMessage(String message) {
        System.out.println("sms fanout ----接收到了订单信息是:-->" + message);
    }
}

@RabbitListener(queues = {"duanxin.fanout.queue"})
@Component
public class FanoutDuanxinConsumer {

    @RabbitHandler
    public void reviseMessage(String message) {
        System.out.println("duanxin fanout ----接收到了订单信息是:-->" + message);
    }
}

运行启动类就可以接收到消息

Direct模式

开发中能用fanout模式就用fanout模式,否则direct模式会消耗性能

config

@Configuration
public class DirectRabbitMQConfiguration {

    // 1:声明注册Direc 模式的交换机
    @Bean
    public DirectExchange directExchange(){
        // 参数名 是否持久化 是否自动删除
        return new DirectExchange("direct_order_exchange",true,false);
    }
    // 2:声明队列 sms.direct.queue duanxin.direct.queue email.direct.queue
    @Bean
    public Queue directsmsQueue(){
        return new Queue("sms.direct.queue",true);
    }

    @Bean
    public Queue directduanxinQueue(){
        return new Queue("duanxin.direct.queue",true);
    }

    @Bean
    public Queue directemailQueue(){
        return new Queue("email.direct.queue",true);
    }
    // 3:完成绑定关系 队列和交换机绑定
    @Bean
    public Binding directsmsBingding(){
        return BindingBuilder.bind(directsmsQueue()).to(directExchange()).with("sms");
    }

    @Bean
    public Binding directduanxinBingding(){
        return BindingBuilder.bind(directduanxinQueue()).to(directExchange()).with("duanxin");
    }

    @Bean
    public Binding directemailBingding(){
        return BindingBuilder.bind(directemailQueue()).to(directExchange()).with("email");
    }
}

生产者

@Service
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 模拟用户下单---direct
     * @param userId
     * @param productId
     * @param num
     */
    public void makeOrderDirect(String userId,String productId,int num){
        // 1:根据商品id查询库存是否充足
        // 2:保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功:"+orderId);
        // 3:通过MQ来完成消息的分发
        // 参数1:交换机  参数2:路由Key/queue队列名称 参数3:消息内容
        String exchangName = "direct_order_exchange";
        rabbitTemplate.convertAndSend(exchangName,"email",orderId);
        rabbitTemplate.convertAndSend(exchangName,"duanxin",orderId);
    }
}

测试

@SpringBootTest
class SpringbootOrderRabbitmqProducerApplicationTests {

	@Autowired
	private OrderService orderService;
	@Test
	void testOrderDirect() {
		orderService.makeOrderDirect("1","1",12);
	}
}

消费者

@RabbitListener(queues = {"sms.direct.queue"})
@Component
public class DirectSMSConsumer {

    @RabbitHandler
    public void reviseMessage(String message) {
        System.out.println("sms direct ----接收到了订单信息是:-->" + message);
    }
}

@RabbitListener(queues = {"email.direct.queue"})
@Component
public class DirectEmailConsumer {

    @RabbitHandler
    public void reviseMessage(String message) {
        System.out.println("email direct ----接收到了订单信息是:-->" + message);
    }
}

@RabbitListener(queues = {"duanxin.direct.queue"})
@Component
public class DirectDuanxinConsumer {

    @RabbitHandler
    public void reviseMessage(String message) {
        System.out.println("duanxin direct ----接收到了订单信息是:-->" + message);
    }
}

面试题:绑定关系配置在生产者好还是配置在消费者好?
其实无论配置在哪一端都是可以的,两边都配置也可以,最好的还是在消费者配置绑定关系,因为启动过程中,消费者如果队列还没有被声明,就会出错,一定要先把队列声明好,因为消费者是直接和队列打交道的地方,消费者是最先起来的服务

topic模式(注解方式的绑定关系)

配置类和注解都可以确定绑定关系,二者选择其一(推荐配置类方式)

@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "sms.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "#.sms.#"
))
public class TopicDuanxinConsumer {

    @RabbitHandler
    public void reviseMessage(String message) {
        System.out.println("duanxin topic ----接收到了订单信息是:-->" + message);
    }
}

@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "email.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "#.email.#"
))
public class TopicEmailConsumer {

    @RabbitHandler
    public void reviseMessage(String message) {
        System.out.println("email topic ----接收到了订单信息是:-->" + message);
    }
}

@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "sms.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "com.#"
))
public class TopicSMSConsumer {

    @RabbitHandler
    public void reviseMessage(String message) {
        System.out.println("sms topic ----接收到了订单信息是:-->" + message);
    }
}

生产者

/**
     * 模拟用户下单---topic
     * @param userId
     * @param productId
     * @param num
     */
    public void makeOrderTopic(String userId,String productId,int num){
        // 1:根据商品id查询库存是否充足
        // 2:保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功:"+orderId);
        // 3:通过MQ来完成消息的分发
        // 参数1:交换机  参数2:路由Key/queue队列名称 参数3:消息内容
        String exchangName = "topic_order_exchange";
        String routingKey="com.sms";
        rabbitTemplate.convertAndSend(exchangName,routingKey,orderId);
    }

RabbitMQ高级

过期时间TTL

概述

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

• 通过队列属性设置,队列中所有消息都有相同的过期时间
• 对消息进行单独设置,每条消息TTL可以不同

如果上述的两种方法同时使用,则消息的过期时间是两者之间TTL较小的那个数值为准.
消息在队列的生存时间一旦超过设置的TTL值,就称为dead message 被投递到死信队列,消费者将无法再收到该消息

设置队列TTL

config

@Configuration
public class TTLRabbitMQConfiguration {

    // 1:声明注册Direc 模式的交换机
    @Bean
    public DirectExchange TTLdirectExchange() {
        // 参数名 是否持久化 是否自动删除
        return new DirectExchange("ttl_direct_exchange", true, false);
    }

       // 2:声明队列 队列的过期时间
    @Bean
    public Queue ttlQueue() {
        // 设置过期时间
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 5000);// 这里一定是个int类型 单位:毫秒
        // 名字 持久化  排他性  自动删除  过期时间参数
        return new Queue("ttl.direct.queue", true,false,false,args);
    }

    // 3:完成绑定关系 队列和交换机绑定
    @Bean
    public Binding ttlBingding() {
        return BindingBuilder.bind(ttlQueue()).to(TTLdirectExchange()).with("ttl");
    }

}

生产者

 /**
     * 模拟用户下单---ttl
     * @param userId
     * @param productId
     * @param num
     */
    public void makeOrderTtl(String userId,String productId,int num){
        // 1:根据商品id查询库存是否充足
        // 2:保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功:"+orderId);
        // 3:通过MQ来完成消息的分发
        // 参数1:交换机  参数2:路由Key/queue队列名称 参数3:消息内容
        String exchangName = "ttl_direct_exchange";
        String routingKey="ttl";
        rabbitTemplate.convertAndSend(exchangName,routingKey,orderId);
    }

测试

@Test
	void testOrderTTl() {
		orderService.makeOrderTtl("1","1",12);
	}

过期时间的信息,一般会用死信队列进行接收

设置消息过期时间TTL

config

@Configuration
public class TTLRabbitMQConfiguration {

    // 1:声明注册Direc 模式的交换机
    @Bean
    public DirectExchange TTLdirectExchange() {
        // 参数名 是否持久化 是否自动删除
        return new DirectExchange("ttl_direct_exchange", true, false);
    }

    // 2:声明队列 队列的过期时间
    @Bean
    public Queue ttlQueue() {
        // 设置过期时间
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 5000);// 这里一定是个int类型 单位:毫秒
        // 名字 持久化  排他性  自动删除  过期时间参数
        return new Queue("ttl.direct.queue", true,false,false,args);
    }

    // 2:声明队列 队列的过期时间
    @Bean
    public Queue ttlMessageQueue() {
        return new Queue("ttl.message.direct.queue", true);
    }
    // 3:完成绑定关系 队列和交换机绑定
    @Bean
    public Binding ttlBingding() {
        return BindingBuilder.bind(ttlQueue()).to(TTLdirectExchange()).with("ttl");
    }
    
 @Bean
    public Binding ttlMessageBingding() {
        return BindingBuilder.bind(ttlMessageQueue()).to(TTLdirectExchange()).with("ttlMessage");
    }

}

生产者

/**
     * 模拟用户下单---ttl
     * @param userId
     * @param productId
     * @param num
     */
    public void makeOrderTtlMessage(String userId,String productId,int num){
        // 1:根据商品id查询库存是否充足
        // 2:保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功:"+orderId);
        // 3:通过MQ来完成消息的分发
        // 参数1:交换机  参数2:路由Key/queue队列名称 参数3:消息内容
        String exchangName = "ttl_direct_exchange";
        String routingKey="ttlMessage";
        // 给消息设置过期时间
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                // 这里就是字符串
                message.getMessageProperties().setExpiration("5000");
                message.getMessageProperties().setContentEncoding("UTF-8");
                return message;
            }
        };
        rabbitTemplate.convertAndSend(exchangName,routingKey,orderId,messagePostProcessor);
    }

测试

@Test
	void testOrderTTlMessage() {
		orderService.makeOrderTtlMessage("1","1",12);
	}

过期队列和过期消息的差异
过期消息,过期了就被移除了不存在
过期队列的消息,过期之后会写入到死信队列中(可以进行消息转移)

死信队列

概述

DLX (Dead-Letter_Exahcnge) 死信交换机(死信邮箱),当消息在一个队列中变成死信之后,它能被重新发送到另一个交换机中,这个交换机就是DLX,绑定DLX的队列就称之为死信队列.
消息变成死信,可能由于以下的原因:

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

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

消息过期

config

@Configuration
public class DeadRabbitMQConfiguration {

    // 1:声明注册Direc 模式的交换机
    @Bean
    public DirectExchange deadDiredct() {
        // 参数名 是否持久化 是否自动删除
        return new DirectExchange("dead_direct_exchange", true, false);
    }

    // 2:声明队列
    @Bean
    public Queue deadQueue() {
        return new Queue("dead.direct.queue", true);
    }

    @Bean
    public Binding deadsBingding() {
        return BindingBuilder.bind(deadQueue()).to(deadDiredct()).with("dead");
    }

}


@Configuration
public class TTLRabbitMQConfiguration {

    // 1:声明注册Direc 模式的交换机
    @Bean
    public DirectExchange TTLdirectExchange() {
        // 参数名 是否持久化 是否自动删除
        return new DirectExchange("ttl_direct_exchange", true, false);
    }

    // 2:声明队列 队列的过期时间
    @Bean
    public Queue ttlQueue() {
        // 设置过期时间
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 5000);// 这里一定是个int类型 单位:毫秒
        args.put("x-dead-letter-exchange","dead_direct_exchange");
        // direct 模式需要设置 key (fanout模式可以不用设置)
        args.put("x-dead-letter-routing-key","dead");
        // 名字 持久化  排他性  自动删除  过期时间参数
        return new Queue("ttl.direct.queue", true,false,false,args);
    }

    // 2:声明队列 队列的过期时间
    @Bean
    public Queue ttlMessageQueue() {
        return new Queue("ttl.message.direct.queue", true);
    }
    // 3:完成绑定关系 队列和交换机绑定
    @Bean
    public Binding ttlBingding() {
        return BindingBuilder.bind(ttlQueue()).to(TTLdirectExchange()).with("ttl");
    }

    @Bean
    public Binding ttlMessageBingding() {
        return BindingBuilder.bind(ttlMessageQueue()).to(TTLdirectExchange()).with("ttlMessage");
    }

}

测试
(修改队列:已存在的队列不会修改和覆盖,只会报错,需要删除掉再重新创建,但真正开发过程中,应该从新创建一个队列不应该删除)

	@Test
	void testOrderTTl() {
		orderService.makeOrderTtl("1","1",12);
	}

队列达到最大长度

config

@Configuration
public class DeadRabbitMQConfiguration {

    // 1:声明注册Direc 模式的交换机
    @Bean
    public DirectExchange deadDiredct() {
        // 参数名 是否持久化 是否自动删除
        return new DirectExchange("dead_direct_exchange", true, false);
    }

    // 2:声明队列
    @Bean
    public Queue deadQueue() {
        return new Queue("dead.direct.queue", true);
    }

    @Bean
    public Binding deadsBingding() {
        return BindingBuilder.bind(deadQueue()).to(deadDiredct()).with("dead");
    }

}

@Configuration
public class TTLRabbitMQConfiguration {

    // 1:声明注册Direc 模式的交换机
    @Bean
    public DirectExchange TTLdirectExchange() {
        // 参数名 是否持久化 是否自动删除
        return new DirectExchange("ttl_direct_exchange", true, false);
    }

    // 2:声明队列 队列的过期时间
    @Bean
    public Queue ttlQueue() {
        // 设置过期时间
        Map<String, Object> args = new HashMap<>();
        // args.put("x-message-ttl", 5000);// 这里一定是个int类型 单位:毫秒
        args.put("x-max-length",5);
        args.put("x-dead-letter-exchange","dead_direct_exchange");
        // direct 模式需要设置 key (fanout模式可以不用设置)
        args.put("x-dead-letter-routing-key","dead");
        // 名字 持久化  排他性  自动删除  过期时间参数
        return new Queue("ttl.direct.queue", true,false,false,args);
    }

    // 2:声明队列 队列的过期时间
    @Bean
    public Queue ttlMessageQueue() {
        return new Queue("ttl.message.direct.queue", true);
    }
    // 3:完成绑定关系 队列和交换机绑定
    @Bean
    public Binding ttlBingding() {
        return BindingBuilder.bind(ttlQueue()).to(TTLdirectExchange()).with("ttl");
    }

    @Bean
    public Binding ttlMessageBingding() {
        return BindingBuilder.bind(ttlMessageQueue()).to(TTLdirectExchange()).with("ttlMessage");
    }

}

测试

@Test
	void testOrderTTl() {
		for (int i = 0; i < 11; i++) {
			orderService.makeOrderTtl("1","1",12);
		}
	}

内存磁盘的监控

RabbitMQ的内存警告

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

如下图:

持久化:把内存里面的数据同步到磁盘

rabbitMQ 的内存控制

参考帮助文档:
https://www.rabbitmq.com/configure.html
当出现警告的时候,可以通过配置去修改和调整

命令的方式

rabbitmqctl set_vm_memory_high_watermark <fraction>  相对值
rabbitmqctl set_vm_memory_high_watermark absolute 50MB  绝对值

fraction/value 为内存的阈值.默认情况是0.4(相对值)/2GB(绝对值) 二者选其一,表示:当rabbitMQ的内存超过40%时,就会产生警告并且阻塞所有生产者的连接
通过此命令修改阈值在Broker 重启以后将会失效,通过修改配置文件方式设置的阈值则不会随着重启而消失,单修改了配置文件一样要重启broker才会生效
(相对值最好写:0.4-0.7之间,不建议超过0.7)

分析:
rabbitmqctl set_vm_memory_high_watermark absolute 50MB

配置文件方式 rabbitmq.conf

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

# 默认
# vm_memory_high_watermark.relative = 0.4
# 使用relative 相对值进行设置fraction 建议取值在0.4-0.7之间,不建议超过0.7
vm_memory_high_watermark.relative = 0.6
# 使用absolute 的绝对值的方式,但是是KB\MB\GB对应的命令如下(8GB*0.6=4GB) 
vm_memory_high_watermark.absolute = 2GB 

rabbitMQ 的内存换页

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

默认情况下，内存到达的阈值时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的值）

为什么设置小于1，因为如果设置为1的阈值，内存都达到了极限，再去换页的意义不是很大

rabbitＭＱ的磁盘预警

当磁盘的剩余空间低于确定的阈值时,rabbitMQ同样会阻塞生产者,这样可以避免因非持久化的消息持续换页而耗尽磁盘空间导致服务器崩溃

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

通过命令方式修改:

rabbitmqctl set_disk_free_limit <disk_limit>
rabbitmqctl set_disk_free_limit memory_limit <fraction>

disk_limit :固定单位 KB MB GB
fraction:是相对阈值,建议范围1.0-2.0之间(相对于内存)

通过配置文件配置:

disk_free_limit.relative = 3.0
disk_free_limit.absolute = 50mb

集群

erlang语言天生具备分布性特征，通过同步Erlang 集群

集群搭建

配置的前提是运行rabbitmq 可以运行起来，

# 查看进程
ps aux|grep rabbitmq

systemctl status rabbitmq-server

注意：确保rabbitmq 可以运行的，确保完成之后，把单机版的rabbitmq 服务停止，后台看不到rabbitmq的进程为止

systemctl stop rabbitmq-server

单机多实例搭建

场景：假设有两个rabbitmq节点，分别是rabbit-1,rabbit-2,rabbit-1作为主节点，rabbit-2作为从节点
启动命令：RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit-1 rabbitmq-server -detached
结束命令：rabbitmqctl -n rabbit-1 stop
具体请查看视频

第一步：启动第一个节点 rabbit-1

sudo RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit-1 rabbitmq-server start &

至此节点rabbit-1启动完成

第二步：启动第二个节点 rabbit-2

注意：web管理插件端口占用，所以还要指定其web插件占用的端口号
RABBITMQ_SERVER_START_ARGS=" -rabbitmq_managment listener [{port,15673}]"

sudo RABBITMQ_NODE_PORT=5673 RABBITMQ_SERVER_START_ARGS=" -rabbitmq_managment listener [{port,15673}]" RABBITMQ_NODENAME=rabbit-2 rabbitmq-server start &

至此节点rabbit-2启动完成

第三步：验证启动 ps aux|grep rabbitmq

第四步：rabbit-1 操作作为主节点

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

rabbit-2操作作为从节点

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

验证集群状态

sudo rabbitmqctl cluster_status -n rabbit-1

web 监控

#安装插件
rabbitmq-plugins enable rabbitmq-management

在访问时需要给15672 node-1 和15673 node-2 设置用户名和密码

rabbitmqctl -n rabbit-1 add_user admin admin 
rabbitmqctl -n rabbit-1 set_user_tags admin administrator
rabbitmqctl -n rabbit-1 set_permissions -p / admin ".*" ".*" ".*"

rabbitmqctl -n rabbit-2 add_user admin admin 
rabbitmqctl -n rabbit-2 set_user_tags admin administrator
rabbitmqctl -n rabbit-2 set_permissions -p / admin ".*" ".*" ".*"

分布式事务

简述

分布式事务是指事物的操作位于不同的节点上，需要保证事务的AICD特性
（例如：在下单场景下，库存和订单如果不在同一个节点上，就涉及分布式事务）

分布式事务的方式

在分布式系统中，要实现分布式事务，有几种解决方案

两阶段提交（2PC）需要数据库产商的支持，Java组件有atomikos等

【学相伴】RabbitMQ最新完整教程IDEA版通俗易懂 | KuangStudy | 狂神说 | 学相伴飞哥https://www.bilibili.com/video/BV1dX4y1V73G?p=37&spm_id_from=pageDriver

内容总结

以上是互联网集市为您收集整理的2021-04-03 狂神说Java之 rabbitmq学习全部内容，希望文章能够帮你解决2021-04-03 狂神说Java之 rabbitmq学习所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---