RabbitMQ学习笔记

一、概述

1、中间件概述

中间件是介于应用系统和系统软件之间的一类软件，它使用系统软件所提供的基础服务（功能），衔接网络上应用系统的各个部分或不同的应用，能够达到资源共享、功能共享的目的。

2、消息中间件概述

1、概述

​ 消息中间件是利用可靠的消息传递机制进行系统和系统直接的通讯；以及通过提供消息传递和消息的排队机制，它可以在分布式系统环境下扩展进程间的通讯。常见的消息中间件有ActiveMQ、RabbitMQ、Kafka、RocketMQ等。

2、应用场景

• 跨系统数据传递
• 高并发的流量削峰
• 数据的分发和异步处理
• 大数据分析与传递
• 分布式事务

3、核心组成部分

• 消息的协议
• 消息的持久化机制
• 消息的分发策略
• 消息的高可用，高可靠
• 消息的容错机制

4、其他

​ 常见的持久化方式

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

​ 消息分发策略的机制和对比

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

二、RabbitMQ安装与入门

1、概述

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

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

2、RabbitMQ安装

1、环境准备

RabbitMQ是采用Erlang语言开发的，所以系统环境必须提供Erlang环境，第一步就是安装Erlang。
RabbitMQ下载地址：https://www.rabbitmq.com/download.html
Erlang下载地址：https://www.erlang-solutions.com/downloads/
erlang和RabbitMQ版本的按照比较: https://www.rabbitmq.com/which-erlang.html

Linux环境

#命令
lsb_release -a
#结果
LSB Version:    :core-4.1-amd64:core-4.1-noarch
Distributor ID: CentOS
Description:    CentOS Linux release 8.1.1911 (Core) 
Release:        8.1.1911
Codename:       Core

2、Erlang安装

rpm包下载地址

#下载rpm包
wget https://packages.erlang-solutions.com/erlang-solutions-2.0-1.noarch.rpm
#更新本地yum仓库
rpm -Uvh erlang-solutions-2.0-1.noarch.rpm
yum install -y erlang
#查看版本测试是否安装成功
erl -v
#一个小插件，需要
yum install -y socat

3、RabbitMQ安装

#下载rpm包
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.9.5/rabbitmq-server-3.9.5-1.el8.noarch.rpm
rpm -Uvh rabbitmq-server-3.9.5-1.el8.noarch.rpm
yum install rabbitmq-server -y

4、RabbitMQ运行

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

5、MQ的相关端口

​ 注意端口的开放与安全组端口开放

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

3、RabbitMQWeb管理界面及授权操作

1、RabbitMQ管理界面安装

#默认情况下，rabbitmq是没有安装web端的客户端插件，需要安装才可以生效
rabbitmq-plugins enable rabbitmq_management
#安装完毕以后，重启服务即可
systemctl restart rabbitmq-server
#注意端口开放
firewall-cmd --zone=public --add-port=15672/tcp --permanent 
firewall-cmd --reload

最后访问http://ip:15672/即可

2、授权账号和密码

​ rabbitmq有一个默认账号和密码是：guest 默认情况只能在localhost本机下访问，所以需要添加一个远程登录的用户。

#新增用户
rabbitmqctl add_user admin admin
#设置用户分配操作权限
rabbitmqctl set_user_tags admin administrator
#为用户添加资源权限
rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

相关操作

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

4、Docker安装RabbitMQ

1、Docker安装

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

详情查看https://blog.csdn.net/lemon_TT/article/details/117983127

2、docker相关命令

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

3、RabbitMQ安装运行

#这个镜像带管理界面
docker pull rabbitmq:management
#运行docker容器
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
#开放端口
firewall-cmd --zone=public --add-port=15672/tcp --add-port=5672/tcp  --permanent
firewall-cmd --reload 
#查看日志
docker logs -f myrabbit
#查看端口是否被占用
netstat -naop | grep 5672

最后访问http://ip:15672/即可

5、RabbitMQ角色介绍

1、none

• 不能访问management plugin

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

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

3、Policymaker

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

4、Monitoring

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

5、Administrator

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

三、RabbitMQ入门实战

1、概述

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

AMQP生产者流转过程

AMQP消费者流转过程

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、Simple简单模式

• 特点：普通消费者生产者模型

1、代码实现

实现步骤

1、jdk1.8
2、构建一个maven工程
3、导入rabbitmq的maven依赖
4、启动rabbitmq-server服务
5、定义生产者
6、定义消费者
7、观察消息的在rabbitmq-server服务中的过程

导入maven依赖

java原生依赖

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

spring依赖

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

springboot依赖

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

上面依赖根据自己的项目环境进行选择即可。

定义生产者

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

定义生产者

public class Comsumer {
    public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("120.24.70.136");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;
        try {
            connection = connectionFactory.newConnection("消费者");
            channel = connection.createChannel();
            channel.basicConsume("queue1", true, new DeliverCallback() {
                @Override
                public void handle(String comsumerTag, Delivery message) throws IOException {
                    System.out.println("收到消息为" + new String(message.getBody(), StandardCharsets.UTF_8));
                }
            }, new CancelCallback() {
                @Override
                public void handle(String s) throws IOException {
                    System.out.println("接受失败了");
                }
            });
            System.out.println("接受消息完毕");
            //System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("接受消息出现异常");
        } finally {
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

2、Web界面操作

3、work工作模式

1、概述

• 特点：分发机制。包括轮询模式和公平分发模式

2、代码实现

1、轮询模式

生产者

public class Producer {
    public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("120.24.70.136");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4: 从连接中获取通道channel
            channel = connection.createChannel();

            // 5： 准备发送消息的内容
            //===============================end topic模式==================================
            for (int i = 1; i <= 20; i++) {
                //消息的内容
                String msg = "shawn:" + i;
                // 7: 发送消息给中间件rabbitmq-server
                // @params1: 交换机exchange
                // @params2: 队列名称/routingkey
                // @params3: 属性配置
                // @params4: 发送消息的内容
                channel.basicPublish("", "queue1", null, msg.getBytes());
            }
            System.out.println("消息发送成功!");
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("发送消息出现异常...");
        } finally {
            // 7: 释放连接关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}

消费者，这里需要创建两个类，work1和work2，内容一样，只贴出一份

public class Work1 {
    public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("120.24.70.136");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;

        try {
            connection = connectionFactory.newConnection("消费者");
            channel = connection.createChannel();
            channel.basicConsume("queue1", true, new DeliverCallback() {
                @Override
                public void handle(String comsumerTag, Delivery message) throws IOException {
                    try {
                        System.out.println("收到消息为" + new String(message.getBody(), StandardCharsets.UTF_8));
                        Thread.sleep(1000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, new CancelCallback() {
                @Override
                public void handle(String s) throws IOException {
                    System.out.println("接受失败了");
                }
            });
            System.in.read();
            System.out.println("接受消息完毕");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("接受消息出现异常");
        } finally {
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

运行结果为两个消费分别消费一个，轮着消费，即使每个线程运行时间不同

2、公平分发模式

生产者不变，消费者改成手动消费，仍然创建两个各，其中每个类运行睡眠时间不一样，结果可以发现对于睡眠时间短的消费消息更多，即性能好的消费更多

package com.shawn.rabbitmq.work.fair;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class Work1 {
    public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("120.24.70.136");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;

        try {
            connection = connectionFactory.newConnection("消费者");
            channel = connection.createChannel();
            Channel finalChannel = channel;
            //表示每次从队列获取一个消息，根据内存状况调整
            finalChannel.basicQos(1);
			//设置手动确认消费消息
            channel.basicConsume("queue1", false, new DeliverCallback() {
                @Override
                public void handle(String comsumerTag, Delivery message) throws IOException {
                    try {
                        System.out.println("收到消息为" + new String(message.getBody(), StandardCharsets.UTF_8));
                        Thread.sleep(200);
                       //确认消费
                     finalChannel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }
            }, new CancelCallback() {
                @Override
                public void handle(String s) throws IOException {
                    System.out.println("接受失败了");
                }
            });
            System.in.read();
            System.out.println("接受消息完毕");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("接受消息出现异常");
        } finally {
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

4、fanout发布订阅模式

1、概述

• 特点：Fanout—发布与订阅模式，是一种广播机制，它是没有路由key的模式。只要订阅了该交换机，那么就可以获取数据

2、代码实现

生产者（注意已经绑定好关系可以不用在代码中编写绑定关系了）

public class Producer {
    public static void main(String[] args) {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("120.24.70.136");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;
        try {
            // 3: 从连接工厂中获取连接
            connection = connectionFactory.newConnection("生产者");
            // 4: 从连接中获取通道channel
            channel = connection.createChannel();
            // 5: 准备交换机,这里我在web界面以及创建完毕
            String exchangeName = "fanout_exchange";
            // 6： 准备发送消息的内容
            String message = "你好，shawn！！！";
            // 7:定义路由key
            String routingKey="";
            // 8: 指定交换机类型
            String type = "fanout";
            //这里队列声明以及绑定关系都已经在web端完成,可以不下以下声明
            //最后true表示重启后交换机不会删除
            channel.exchangeDeclare(exchangeName, type, true);
            String queueName1 = channel.queueDeclare("queue1", false, false, false, null).getQueue();
            channel.queueBind(queueName1, exchangeName, routingKey);
            String queueName2 = channel.queueDeclare("queue2", false, false, false, null).getQueue();
            channel.queueBind(queueName2, exchangeName, routingKey);
            // 9: 发送消息给中间件rabbitmq-server
            // @params1: 交换机exchange
            // @params2: 队列名称/routing
            // @params3: 属性配置
            // @params4: 发送消息的内容
            channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
            System.out.println("消息发送成功!");
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("发送消息出现异常...");
        } finally {
            // 7: 释放连接关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}

消费者

//这里注意消费者会将消息都消费
public class Comsumer {
     private static final Runnable runnable = () -> {
        // 1: 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2: 设置连接属性
        connectionFactory.setHost("120.24.70.136");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = null;
        Channel channel = null;
         //获取队列的名称
         final String queueName = Thread.currentThread().getName();

        try {
            connection = connectionFactory.newConnection("消费者");
            channel = connection.createChannel();
            channel.basicConsume(queueName, true, new DeliverCallback() {
                @Override
                public void handle(String comsumerTag, Delivery message) throws IOException {
                    System.out.println("收到消息为" + new String(message.getBody(), StandardCharsets.UTF_8));
                }
            }, new CancelCallback() {
                @Override
                public void handle(String s) throws IOException {
                    System.out.println("接受失败了");
                }
            });
            System.in.read();
            System.out.println("接受消息完毕");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("接受消息出现异常");
        } finally {
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    };
    public static void main(String[] args){
        new Thread(runnable,"queue1").start();
        new Thread(runnable,"queue2").start();
    }
}

5、direct路由模式

• 特点：有routing-key的匹配模式，根据给定的路由key确定对应的队列

​ 代码实现如fanout模式，将生产者交换机以及路由修改即可，这里路由是精确匹配

6、topic主题模式

• 特点：模糊的routing-key的匹配模式

  #表示0级或多级

  *表示必须有1级

​ 代码实现如fanout模式，将生产者交换机以及路由修改即可，这里路由是正则匹配方式

7、headers参数模式

• 特点：通过参数进行相应队列的分发

四、SPringBoot整合RabbitMQ入门实战

1、环境准备

​ 创建SPringBoot父子项目，兵在pom.xml引入RabbitMQ依赖，后面就创建子模块，pom依赖关系配置详见父子项目搭建参考

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

​ 目标架构

2、fanout模式

1、创建生产者模块

创建springboot-rabbitmq-fanout-producer模块，并在在application.yml进行配置

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

定义订单生产者

package com.shawn.springbootrabbitmqfanoutproducer.service;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.UUID;

@Service
public class OrderService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 1: 定义交换机
    private String exchangeName = "fanout_order_exchange";
    // 2: 路由key
    private String routeKey = "";

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

定义绑定关系,，相当于之前的创建交换机、绑定队列等

package com.shawn.springbootrabbitmqfanoutproducer.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author Shawn
 * @date 2021/09/02
 */
@Configuration
public class FanoutConfiguration {

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

}
}

最后进行测试类，成功后即可发现队列有了消息

@SpringBootTest
class SpringbootRabbitmqFanoutProducerApplicationTests {
    @Autowired
    OrderService orderService;

    @Test
    void test1() throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            Thread.sleep(1000);
            Long userId = 100L + i;
            Long productId = 10001L + i;
            int num = 10;
            orderService.makeOrder(userId, productId, num);
        }
    }
}

2、创建消费者模块

创建springboot-rabbitmq-fanout-consumer模块，并在在application.yml进行配置

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

分别创建三个消费者类，这里举例其中一个SMS消费类，最后启动即可

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

3、direct模式

​ 新建项目，配置DirectConfiguration中的交换机和队列，并在订单业务中设置路由key，其他操作与fanout模式类似

@Configuration
public class DirectConfiguration {

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

4、topic模式

​ 在这里可以不定义figuration，直接通过队列绑定交换机的路由关系，其他操作与fanout模式类似（通过注解方式绑定，之前都是通过配置文件方式绑定）

// bindings其实就是用来确定队列和交换机绑定关系
@RabbitListener(bindings =@QueueBinding(
        // sms.topic.queue 是队列名字，这个名字你可以自定随便定义。
        value = @Queue(value = "sms.topic.queue",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",
                type = ExchangeTypes.TOPIC),
        key = "#.sms.#"
))
@Component
public class TopicSMSConsumer {

    @RabbitHandler
    public void messagerevice(String message){
        // 此处省略发邮件的逻辑
        System.out.println("SMS-------------->" + message);
    }
}

5、过期时间ttl和死信队列(★)

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

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

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

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

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

​ 首先定义交换机、队列以及绑定关系，在定义队列时设置ttl参数以及dead参数，即创建后该队列会显示ttl以及DLX，代表是会自动删除消息的队列，并且删除的消息会发送到死信队列。这里定义了两种过期类型：队列过期和消息过期，其中队列过期可以将过期消息送到死信队列，而消息过期是完全删除了消息。

@Configuration
public class TTLDirectConfiguration {

    //队列的过期时间
    @Bean
    public Queue ttlQueue(){
        Map<String,Object> args = new HashMap<>();
        // 这个key可以在图形化界面查看，类型一定是int类型
        args.put("x-message-ttl",5000);
        args.put("x-dead-letter-exchange","dead_direct_exchange");
        // fanout模式不用配置
        args.put("x-dead-letter-routing-key","dead");
        return new Queue("ttl.direct.queue",true,false,false,args);
    }
    @Bean
    public Queue ttlMessageQueue(){
        return new Queue("ttl.message.direct.queue",true);
    }

    //交换机名字和类型
    @Bean
    public DirectExchange TTLDirectExchange() {
        return new DirectExchange("ttl_direct_exchange", true, false);
    }

    @Bean
    public Binding directTTLBinding(){
        return BindingBuilder.bind(ttlQueue()).to(TTLDirectExchange()).with("ttl");
    }

    @Bean
    public Binding directTTLBinding1(){
        return BindingBuilder.bind(ttlMessageQueue()).to(TTLDirectExchange()).with("ttl.message");
    }
}

//死信队列交换机和队列配置
@Configuration
public class DeadDirectConfiguration {

    @Bean
    public Queue deadQueue(){
        return new Queue("dead.direct.queue",true);
    }
    //交换机名字和类型
    @Bean
    public DirectExchange deadDirectExchange() {
        return new DirectExchange("dead_direct_exchange", true, false);
    }

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

​ 创建服务类

@Service
public class TTLService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 1: 定义交换机
    private String exchangeName = "ttl_direct_exchange";

    // 队列设置过期
    public void makeOrder(Long userId, Long productId, int num) {
        String routeKey = "ttl";
        // 1： 模拟用户下单
        String orderNumer = UUID.randomUUID().toString();
        System.out.println("用户 " + userId + ",订单编号是：" + orderNumer);
        // 发送订单信息给RabbitMQ
        rabbitTemplate.convertAndSend(exchangeName, routeKey, orderNumer);
    }

    // 消息设置过期
    public void makeOrderMessage(Long userId, Long productId, int num) {
        String routeKey = "ttl.message";
        // 1： 模拟用户下单
        String orderNumer = UUID.randomUUID().toString();
        System.out.println("用户 " + userId + ",订单编号是：" + orderNumer);
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //设置5s过期
                message.getMessageProperties().setExpiration("5000");
                message.getMessageProperties().setContentEncoding("utf-8");
                return message;
            }
        };
        // 发送订单信息给RabbitMQ
        rabbitTemplate.convertAndSend(exchangeName, routeKey, orderNumer,messagePostProcessor);
    }
}

​ 最后创建测试类，启动后过期队列的消息过期会送到死信队列，而过期消息则直接删除

@SpringBootTest
class SpringbootRabbitmqFanoutProducerApplicationTests {
    @Autowired
    TTLService ttlService;

    @Test
    void test4() throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            Thread.sleep(1000);
            Long userId = 100L + i;
            Long productId = 10001L + i;
            int num = 10;
            // 队列过期
            ttlService.makeOrder(userId, productId, num);
            // 消息过期
            ttlService.makeOrderMessage(userId, productId, num);
        }
    }
}

6、消息确认机制的配置

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

# 服务端口
server:
  port: 8080
# 配置rabbitmq服务
spring:
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    host: 120.24.70.136
    port: 5672
#    集群配置方式
#    addresses: 120.24.70.136:5672
    publisher-confirm-type: correlated

​ 在代码中配置确认机制，生产者无论成功发送与否，都会收到消息

//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) {
                if(ack){
                    System.out.println("消息确认成功!!!!");
                }else{
                    System.out.println("消息确认失败!!!!");
                }
            }
        });
    }

​ 对于消费者来说，需要在配置文件重新配置过，解决消息重试的几种方案：

• 控制重发的次数
• try+catch+手动ack
• try+catch+手动ack+死信队列处理

7、序列化和反序列化

传输bean对象时可能会出现list无法传输，有两种方案

• 始终转换推断类型converter.setAlwaysConvertToInferredType(true);
• 将 spring-amqp 升级到 2.2.13.RELEASE 或以上

/**
 * @author Shawn
 * @date 2021年11月25日15:35
 **/
@Configuration
public class RabbitMQConfig implements RabbitListenerConfigurer {


    // 可以将json串反序列化为对象
    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar rabbitListenerEndpointRegistrar) {
        rabbitListenerEndpointRegistrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
    }

    @Bean
    MessageHandlerMethodFactory messageHandlerMethodFactory(){
        DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();
        messageHandlerMethodFactory.setMessageConverter(mappingJackson2MessageConverter());
        return messageHandlerMethodFactory;
    }

    @Bean
    public MappingJackson2MessageConverter mappingJackson2MessageConverter(){
        return  new MappingJackson2MessageConverter();
    }

    // 提供自定义RabbitTemplate,将对象序列化为json串
    @Bean
    public RabbitTemplate jacksonRabbitTemplate(ConnectionFactory connectionFactory) {

        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
        jackson2JsonMessageConverter.setAlwaysConvertToInferredType(true);
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter);
        return rabbitTemplate;
    }
}

五、RabbitMQ内存磁盘的监控

1、概述

​ 把消息默认放在内存中是为了加快传输和消费的速度，存入磁盘是保证消息数据的持久化。

参考帮助文档：https://www.rabbitmq.com/configure.html

​ 当RabbitMQ警告时，即内存或者磁盘爆红，所有队列会进入阻塞状态，RabbitMQ无法正常运行。当出现警告的时候，可以通过配置去修改和调整

2、RabbitMQ的内存控制

1、命令的方式

​ fraction/value 为内存阈值。默认情况是：0.4/2GB，代表的含义是：当RabbitMQ的内存超过40%时，就会产生警告并且阻塞所有生产者的连接。通过此命令修改阈值在Broker重启以后将会失效，通过修改配置文件方式设置的阈值则不会随着重启而消失，但修改了配置文件一样要重启broker才会生效。

rabbitmqctl set_vm_memory_high_watermark <fraction>
#绝对内存大小
rabbitmqctl set_vm_memory_high_watermark absolute 50MB
#相对内存大小
rabbitmqctl set_vm_memory_high_watermark 0.4

2、配置文件方式 rabbitmq.conf

​ 当前配置文件：/etc/rabbitmq/rabbitmq.conf（若不存在可自行创建）

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

3、RabbitMQ的内存换页

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

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

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

vm_memory_high_watermark.relative = 0.4
#设置小于1的值，大于1就没必要
vm_memory_high_watermark_paging_ratio = 0.7

4、RabbitMQ的磁盘预警

​ 当磁盘的剩余空间低于确定的阈值时，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

六、RabbitMQ高级

1、消息队列高可用和高可靠

所谓高可用：是指产品在规定的条件和规定的时刻或时间内处于可执行规定功能状态的能力。
当业务量增加时，请求也过大，一台消息中间件服务器的会触及硬件（CPU,内存，磁盘）的极限，一台消息服务器你已经无法满足业务的需求，所以消息中间件必须支持集群部署。来达到高可用的目的。

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

1、 Master-slave主从共享数据模式

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

2、Master- slave主从同步模式

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

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

​ 其写入可以往任意节点去写入。

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

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

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

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

2、集群搭建

1、概述

RabbitMQ这款消息队列中间件产品本身是基于Erlang编写，Erlang语言天生具备分布式特性（通过同步Erlang集群各节点的magic cookie来实现）。因此，RabbitMQ天然支持Clustering。这使得RabbitMQ本身不需要像ActiveMQ、Kafka那样通过ZooKeeper分别来实现HA方案和保存集群的元数据。集群是保证可靠性的一种方式，同时可以通过水平扩展以达到增加消息吞吐量能力的目的。
官网参考：https://www.rabbitmq.com/clustering.html

2、环境准备

​ 保证RabbitMQ是可执行的，并把单机版的RabbitMQ服务停止，后台看不到RabbitMQ的进程为止。这里我在一台主机发布多个RabbitMQ，用的是 Master-slave主从共享数据模式。

#查看进程
ps aux|grep rabbitmq
systemctl status rabbitmq-server

3、单机多实例搭建

​ 分别启动两个节点

sudo RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit-1 rabbitmq-server start &
RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit-1 rabbitmq-server -detached
#web管理插件端口占用,所以还要指定其web插件占用的端口号
sudo RABBITMQ_NODE_PORT=5673 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME=rabbit-2 rabbitmq-server start &
RABBITMQ_NODE_PORT=5673 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME=rabbit-2 rabbitmq-server -detached
#验证是否启动
ps aux|grep rabbitmq

​ rabbit-1操作作为主节点

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

​ rabbit2操作为从节点

# 停止应用
sudo rabbitmqctl -n rabbit-2 stop_app
# 目的是清除节点上的历史数据（如果不清除，无法将节点加入到集群）
sudo rabbitmqctl -n rabbit-2 reset
# 将rabbit2节点加入到rabbit1（主节点）集群当中(Server-node服务器的主机名,需要根据自己情况替换)
sudo rabbitmqctl -n rabbit-2 join_cluster rabbit-1@Server-node
rabbitmqctl -n rabbit-2 forget_cluster_node rabbit-1@Server-node
# 启动应用
sudo rabbitmqctl -n rabbit-2 start_app

​ 验证集群状态

sudo rabbitmqctl cluster_status -n rabbit-1
#设置web界面
rabbitmq-plugins enable rabbitmq_management

4、其他

​ 如果采用多机部署方式，需读取其中一个节点的cookie, 并复制到其他节点（节点之间通过cookie确定相互是否可通信）。cookie存放在/var/lib/rabbitmq/.erlang.cookie。
​ 例如：主机名分别为rabbit-1、rabbit-2
​ 1、逐个启动各节点
​ 2、配置各节点的hosts文件( vim /etc/hosts)或者加入主节点的时候采用ip
​ ip1：rabbit-1
​ ip2：rabbit-2
​ 其它步骤雷同单机部署方式，另外对于集群来说，springboot的yml配置需要更改为集群模式连接

3、下单配送分布式高可用实战

4、其他配置详解

1、application.yml详解

rabbitmq:
  addresses: 127.0.0.1:6605,127.0.0.1:6606,127.0.0.1:6705 #指定client连接到的server的地址，多个以逗号分隔(优先取addresses，然后再取host)
  #    port:
  ##集群配置 addresses之间用逗号隔开
  # addresses: ip:port,ip:port
  password: admin
  username: 123456
  virtual-host: / # 连接到rabbitMQ的vhost
  requested-heartbeat: #指定心跳超时，单位秒，0为不指定；默认60s
  publisher-confirms: #是否启用 发布确认
  publisher-reurns: # 是否启用发布返回
  connection-timeout: #连接超时，单位毫秒，0表示无穷大，不超时
  cache:
    channel.size: # 缓存中保持的channel数量
    channel.checkout-timeout: # 当缓存数量被设置时，从缓存中获取一个channel的超时时间，单位毫秒；如果为0，则总是创建一个新channel
    connection.size: # 缓存的连接数，只有是CONNECTION模式时生效
    connection.mode: # 连接工厂缓存模式：CHANNEL 和 CONNECTION
  listener:
    simple.auto-startup: # 是否启动时自动启动容器
    simple.acknowledge-mode: # 表示消息确认方式，其有三种配置方式，分别是none、manual和auto；默认auto
    simple.concurrency: # 最小的消费者数量
    simple.max-concurrency: # 最大的消费者数量
    simple.prefetch: # 指定一个请求能处理多少个消息，如果有事务的话，必须大于等于transaction数量.
    simple.transaction-size: # 指定一个事务处理的消息数量，最好是小于等于prefetch的数量.
    simple.default-requeue-rejected: # 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
    simple.idle-event-interval: # 多少长时间发布空闲容器时间，单位毫秒
    simple.retry.enabled: # 监听重试是否可用
    simple.retry.max-attempts: # 最大重试次数
    simple.retry.initial-interval: # 第一次和第二次尝试发布或传递消息之间的间隔
    simple.retry.multiplier: # 应用于上一重试间隔的乘数
    simple.retry.max-interval: # 最大重试时间间隔
    simple.retry.stateless: # 重试是有状态or无状态
  template:
    mandatory: # 启用强制信息；默认false
    receive-timeout: # receive() 操作的超时时间
    reply-timeout: # sendAndReceive() 操作的超时时间
    retry.enabled: # 发送重试是否可用
    retry.max-attempts: # 最大重试次数
    retry.initial-interval: # 第一次和第二次尝试发布或传递消息之间的间隔
    retry.multiplier: # 应用于上一重试间隔的乘数
    retry.max-interval: #最大重试时间间隔

对于发送方而言，需要做以下配置：

• 配置CachingConnectionFactory

• 配置Exchange/Queue/Binding

• 配置RabbitAdmin创建上一步的Exchange/Queue/Binding

• 配置RabbitTemplate用于发送消息，RabbitTemplate通过CachingConnectionFactory获取到Connection，然后想指定Exchange发送

对于消费方而言，需要做以下配置：

• 配置CachingConnectionFactory
• 配置Exchange/Queue/Binding
• 配置RabbitAdmin创建上一步的Exchange/Queue/Binding
• 配置RabbitListenerContainerFactory
• 配置@RabbitListener/@RabbitHandler用于接收消息

默认情况下主要的配置

Spring AMQP的主要对象

2、通过java bean方式配置

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class RabbitConfig {
    private static final Logger logger = LoggerFactory.getLogger(RabbitConfig.class);
    public static final String RECEIVEDLXEXCHANGE="spring-ex";
    public static final String RECEIVEDLXQUEUE="spring-qu1";
    public static final String RECEIVEDLXROUTINGKEY="aa";
    public static final String DIRECTEXCHANGE="spring-ex";
    public static final String MDMQUEUE="mdmQueue";
    public static final String TOPICEXCHANGE="spring-top";
    @Value("${spring.rabbitmq.addresses}")
    private String hosts;
    @Value("${spring.rabbitmq.username}")
    private String userName;
    @Value("${spring.rabbitmq.password}")
    private String password;
    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;
 /*   @Value("${rabbit.channelCacheSize}")
    private int channelCacheSize;*/
//    @Value("${rabbit.port}")
//    private int port;
/*    @Autowired
    private ConfirmCallBackListener confirmCallBackListener;
    @Autowired
    private ReturnCallBackListener returnCallBackListener;*/
    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setAddresses(hosts);
        cachingConnectionFactory.setUsername(userName);
        cachingConnectionFactory.setPassword(password);
//        cachingConnectionFactory.setChannelCacheSize(channelCacheSize);
        //cachingConnectionFactory.setPort(port);
        cachingConnectionFactory.setVirtualHost(virtualHost);
        //设置连接工厂缓存模式：
        cachingConnectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
        //缓存连接数
        cachingConnectionFactory.setConnectionCacheSize(3);
        //设置连接限制
        cachingConnectionFactory.setConnectionLimit(6);
        logger.info("连接工厂设置完成，连接地址{}"+hosts);
        logger.info("连接工厂设置完成，连接用户{}"+userName);
        return cachingConnectionFactory;
    }
    @Bean
    public RabbitAdmin rabbitAdmin(){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
        rabbitAdmin.setAutoStartup(true);
        rabbitAdmin.setIgnoreDeclarationExceptions(true);
        rabbitAdmin.declareBinding(bindingMdmQueue());
        //声明topic交换器
        rabbitAdmin.declareExchange(directExchange());
        logger.info("管理员设置完成");
        return rabbitAdmin;
    }
    @Bean
    public RabbitListenerContainerFactory listenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //最小消费者数量
        factory.setConcurrentConsumers(10);
        //最大消费者数量
        factory.setMaxConcurrentConsumers(10);
        //一个请求最大处理的消息数量
        factory.setPrefetchCount(10);
        //
        factory.setChannelTransacted(true);
        //默认不排队
        factory.setDefaultRequeueRejected(true);
        //手动确认接收到了消息
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        logger.info("监听者设置完成");
        return factory;
    }
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange(DIRECTEXCHANGE,true,false);
    }
    @Bean
    public Queue mdmQueue(){
        Map arguments = new HashMap<>();
        // 绑定该队列到私信交换机
        arguments.put("x-dead-letter-exchange",RECEIVEDLXEXCHANGE);
        arguments.put("x-dead-letter-routing-key",RECEIVEDLXROUTINGKEY);
        logger.info("队列交换机绑定完成");
        return new Queue(RECEIVEDLXQUEUE,true,false,false,arguments);
    }
    @Bean
    Binding bindingMdmQueue() {
        return BindingBuilder.bind(mdmQueue()).to(directExchange()).with("");
    }
    @Bean
    public RabbitTemplate rabbitTemplate(){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        rabbitTemplate.setMandatory(true);
        //发布确认
//        rabbitTemplate.setConfirmCallback(confirmCallBackListener);
        // 启用发布返回
//        rabbitTemplate.setReturnCallback(returnCallBackListener);
        logger.info("连接模板设置完成");
        return rabbitTemplate;
    }
  /*  @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(TOPICEXCHANGE,true,false);
    }*/
  /*
*//**
     * @return DirectExchange
     *//*
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(RECEIVEDLXEXCHANGE,true,false);
    }
*//*
*
     * @return Queue
*//*
    @Bean
    public Queue dlxQueue() {
        return new Queue(RECEIVEDLXQUEUE,true);
    }
*//*
     * @return Binding
     *//*
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(RECEIVEDLXROUTINGKEY);
    }*/
}

赏

谢谢你请我吃糖果

支付宝