我家小花儿

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

热门标签

RabbitMQ安装配置，超详细版教程

作者：我家小花儿 | 2024-07-06 12:10:19

踩

rabbitmq安装配置

消息队列（MQ）

什么是消息队列

MQ（Message Queue）消息队列，是基础数据结构中“先进先出”的一种数据机构。指把要传输的数据（消息）放在队列中，用队列机制来实现消息传递——生产者产生消息并把消息放入队列，然后由消费者去处理。消费者可以到指定队列拉取消息，或者订阅相应的队列，由MQ服务端给其推送消息
消息队列是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。

结合前面所说的问题：

商品服务对商品增删改以后，无需去操作索引库，只是发送一条消息，也不关心消息被谁接收。
搜索服务服务接收消息，去处理索引库。

如果以后有其它系统也依赖商品服务的数据，同样监听消息即可，商品服务无需任何代码修改。

AMQP 和 JMS

MQ是消息通信的模型，并不是具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。

AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等。
JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。

两者间的区别和联系

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

常见MQ产品

ActiveMQ：基于JMS
RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会
Kafka：分布式消息系统，高吞吐量

RabbitMQ

RabbitMQ是基于AMQP的一款消息管理系统

官网： http://www.rabbitmq.com/
官方教程：http://www.rabbitmq.com/getstarted.html
RabbitMQ 是一个开源的, 在AMQP基础上完整的,可复用的企业消息系统
支持主流的操作系统, Linux, window , MacOS等
多种开发语言支持 : Java, Python, Ruby, .NET, PHP, C/C++, Node.js 等

下载安装

官网下载地址：http://www.rabbitmq.com/download.html
安装 docker: 安装教程

docker 中安装

下载镜像：docker pull rabbitmq:management
创建实例并启动：

docker run -d --name rabbitmq --publish 5671:5671 \ --publish 5672:5672 --publish 4369:4369 --publish 25672:25672 --publish 15671:15671 --publish 15672:15672 \ rabbitmq:management

注：

4369 – erlang 发现口
5672 --client 端通信口
15672 – 管理界面ui端口
25672 – server 间内部通信口

linux中安装

安装Erlang、Socat、RabbitMQ
下载 :https://yonxin.lanzouv.com/b00zkh65e
密码:evu9
安装包到虚拟机 /opt/MQ目录下

按顺序执行安装命令：

rpm -ivh erlang-21.3.8.9-1.el7.x86_64.rpm
rpm -ivh socat-1.7.3.2-1.el6.lux.x86_64.rpm 
rpm -ivh rabbitmq-server-3.8.1-1.el7.noarch.rpm
1
2
3

安装成功后rabbitmq命令存放在：
/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.1/sbin/
在这里插入图片描述

启用管理插件

rabbitmq-plugins enable rabbitmq_management

在这里插入图片描述

RabbitMQ启停命令

systemctl start rabbitmq-server.service 	#启动

systemctl status rabbitmq-server.service    #查看状态

systemctl restart rabbitmq-server.service   #重启

systemctl stop rabbitmq-server.service 	    #停止
1
2
3
4
5
6
7

查看进程 : ps -ef | grep rabbitmq

测试

启动成功后查看占用的端口 : netstat -anp
在web浏览器中输入地址：http://虚拟机ip:15672/
输入默认账号: guest : guest，默认不允许远程连接
增加管理员账号：rabbitmqctl add_user admin admin
给账号分配角色：rabbitmqctl set_user_tags admin administrator
使用刚才创建好的角色登录, 进入首页
overview：概览
connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况
channels：通道，建立连接后，会形成通道，消息的投递获取依赖通道。
Exchanges：交换机，用来实现消息的路由
Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列。
修改角色密码：rabbitmqctl change_password admin 123456
查看用户列表：rabbitmqctl list_users

端口：

5672 : rabbitMq的编程语言客户端连接端口

15672：rabbitMq管理界面端口

25672：rabbitMq集群的端口

管理界面

管理界面添加用户
通过管理界面我们也可以自己创建一个用户：

超级管理员(administrator)

可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。
1

监控者(monitoring)

可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
1

策略制定者(policymaker)

可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。
1

普通管理者(management)

仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。
1

其他

无法登陆管理控制台，通常就是普通的生产者和消费者。
1

创建Virtual Hosts

虚拟主机：类似于mysql中的database。他们都是以“/”开头
添加完成后默认会给当前登录的用户
点击虚拟主机, 分配多用户使用

五种消息模型

RabbitMQ提供了6种消息模型，但是第6种其实是RPC，并不是MQ，因此不予学习。那么也就剩下5种。

但是其实3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。
在这里插入图片描述

第一种 : 类似于 redis 的 list, 一边推消息, 一边获取消息, 支持多种语言, 是一个基本的消息模型 , 一对一的消息模型, 一个生产者对应一个消费者
第二种 : 工作消息模型如果里面的消息比较多, 需要争抢这个消息, 可以使用工作的消息模型, 使通过一个生产者去推消息, 多个消费之去争抢这个消息去消费, 都会从这个消息队列里面去获取2
第三种 : 发布订阅的模式通过一个生产者向交换机里面推消息 , 推消息时是通过交换机去推, 交换机通过消息的类型把它发布到不同的消息队列里面, 然后消费者根据指定的消息队列,去获取不同消息队列里面的这个消息
第四种 : 也称为广播。就是把生产者把数据交给我们的交换机, 然后根据不同类型广播到指定的队列里面, 消费者就绑定指定类型的消息队列,从这里面去取数据, 根据不同的数据发送数据, 只会推送去数据,但不关心是否成功
第五种 : 订阅模型, 生产者把消息交给交换机, 交换机在把消息推送到队列里面, 将数据推出去后,需要考虑返回的结果, 有没有接收成功, 防止数据丢失

代码连接rabbitmq

创建springboot demo
在pom.xml文件中引入依赖

<dependency>
  <groupId>org.apache.commons</groupId>
  <artifactId>commons-lang3</artifactId>
  <version>3.3.2</version>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4
5
6
7
8
9

编写 ConnectionUtil.java 工具类

package com.rabbitmq.utils;

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

/**
 * 获取 Rabbitmq 连接的工具类
 * Created by YongXin Xue on 2020/06/11 21:06
 */
public class ConnectionUtil {

    /**
     * 获取连接对象
     * @return
     */
    public static Connection getConnection() throws Exception {
        // 创建连接工厂对象
        ConnectionFactory factory = new ConnectionFactory();
        // 给工厂对象设置参数: rabbitmq服务的参数[IP, 端口号, 账号密码, 虚拟主机]
        factory.setHost("192.168.230.205");     // 服务器IP
        factory.setPort(5672);      // 端口号
        // 设置连接超时的时间
        factory.setConnectionTimeout(3000);
        // 设置账号密码, 创建的虚拟机名
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/host_admin");
        // 使用工厂对象,创建连接对象
        Connection connection = factory.newConnection();
        return connection;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

在测试类中测试

@Test
void contextLoads() throws Exception {
    // 测试获取 rabbitmq 连接
    Connection connection = ConnectionUtil.getConnection();
    System.out.println("connection = " + connection);
}
1
2
3
4
5
6

结果 : connection = amqp://admin@192.168.230.205:5672//host_admin
1

基本消息模型-simple

官方介绍：

RabbitMQ是一个消息代理：它接受和转发消息。你可以把它想象成一个邮局：当你把邮件放在邮箱里时，你可以确定邮差先生最终会把邮件发送给你的收件人。在这个比喻中，RabbitMQ是邮政信箱，邮局和邮递员。
RabbitMQ与邮局的主要区别是它不处理纸张，而是接受，存储和转发数据消息的二进制数据块。
P（producer/ publisher）：生产者，一个发送消息的用户应用程序。
C（consumer） ：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序
队列（红色区域）：rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序，但是它们只能存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。

总之：

生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。
我们将用Java编写两个程序;发送单个消息的生产者，以及接收消息并将其打印出来的消费者。我们将详细介绍Java API中的一些细节，这是一个消息传递的“Hello World”。
我们将调用我们的消息发布者（发送者）Send和我们的消息消费者（接收者）Recv。发布者将连接到RabbitMQ，发送一条消息，然后退出。

订阅模型分类

订阅模型-Fanout

Fanout，也称为广播。
在广播模式下，消息发送流程是这样的：

1）可以有多个消费者
2）每个消费者有自己的queue（队列）
3）每个队列都要绑定到Exchange（交换机）
4） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
5）交换机把消息发送给绑定过的所有队列
6）队列的消费者都能拿到消息。实现一条消息被多个消费者消费

生产者

两个变化：

1）声明Exchange，不再声明Queue
2）发送消息到Exchange，不再发送到Queue

package com.rabbitmq.fanout;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.utils.ConnectionUtil;

/**
 * 第一种 : 广播模式的生产者
 * Created by YongXin Xue on 2020/06/11 21:52
 */
public class Send {
    private static final String EXCHANGE_NAME = "fanout_exchange";
    private static final String MSG = "Hi 来自广播模式的一条新消息 !";
    public static void main(String[] args) throws Exception {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 建立通道
        Channel channel = connection.createChannel();
        // 声明交换机, 并指定类型
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        // 使用通道将消息发布到交换机 参数1: 指定交换机, 如果使用空字符串,表会使用默认的交换机
        channel.basicPublish(EXCHANGE_NAME, "", null, MSG.getBytes());
        // 关闭通道
        channel.close();
        // 关闭连接
        connection.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

消费者1

package com.rabbitmq.fanout;

import com.rabbitmq.client.*;
import com.rabbitmq.utils.ConnectionUtil;

import java.io.IOException;

/**
 * 广播模式的消费者 1
 * Created by YongXin Xue on 2020/06/11 22:12
 */
public class Consumer_1 {
    private static final String RECV1_QUEUE_NAME = "recv1_fanout_queue";
    // 交换机名称
    private static final String EXCHANGE_NAME = "fanout_exchange";
    public static void main(String[] args) throws Exception {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列 : 生产者 -> 交换机 -> 队列 ->消费者
        channel.queueDeclare(RECV1_QUEUE_NAME,false, false, false, null);
        // 绑定队列到交换机
        // 参数1: 队列名称 参数2: 交换机名称 参数3: routing Key
        channel.queueBind(RECV1_QUEUE_NAME, EXCHANGE_NAME, "");
        // 创建消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * 接收到数据的回调方法
             * @param consumerTag
             * @param envelope
             * @param properties
             * @param body  生产者发送过来的数据的字节数组
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("[广播模式的消费者 1 = " + msg + " ]");
            }
        };
        // 阻塞监听队列 true 设置自动对回执
        channel.basicConsume(RECV1_QUEUE_NAME, true, consumer);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

消费者2

package com.rabbitmq.fanout;

import com.rabbitmq.client.*;
import com.rabbitmq.utils.ConnectionUtil;

import java.io.IOException;

/**
 * 广播模式的消费者 2
 * Created by YongXin Xue on 2020/06/11 22:12
 */
public class Consumer_2 {
    private static final String RECV2_QUEUE_NAME = "recv2_fanout_queue";
    // 交换机名称
    private static final String EXCHANGE_NAME = "fanout_exchange";
    public static void main(String[] args) throws Exception {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列 : 生产者 -> 交换机 -> 队列 ->消费者
        channel.queueDeclare(RECV2_QUEUE_NAME, false, false, false, null);
        // 绑定队列到交换机
        // 参数1: 队列名称 参数2: 交换机名称 参数3: routing Key
        channel.queueBind(RECV2_QUEUE_NAME, EXCHANGE_NAME, "");
        // 创建消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * 接收到数据的回调方法
             * @param consumerTag
             * @param envelope
             * @param properties
             * @param body  生产者发送过来的数据的字节数组
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("[广播模式的消费者 2 = " + msg + " ]");
            }
        };
        // 阻塞监听队列 true 设置自动对回执
        channel.basicConsume(RECV2_QUEUE_NAME, true, consumer);
    }
}

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

测试

我们运行两个消费者，然后发送1条消息：
在这里插入图片描述

应用场景：文字直播

订阅模型-Direct

有选择性的接收消息

在订阅模式中，生产者发布消息，所有消费者都可以获取所有消息。
在路由模式中，我们将添加一个功能 - 我们将只能订阅一部分消息。例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。
但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
息的发送方在向Exchange发送消息时，也必须指定消息的routing key。
P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

生产者

此处我们模拟商品的增删改，发送消息的RoutingKey分别是：insert、update、delete

package com.rabbitmq.direct;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.utils.ConnectionUtil;

/**
 * direct 模式
 * Created by YongXin Xue on 2020/06/11 23:17
 */
public class Send {
    private static final String EXCHANGE_NAME = "direct_exchange";
    private static final String MSG = "[error]: java.lang.NullPointerException";
    public static void main(String[] args) throws Exception {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 建立通道
        Channel channel = connection.createChannel();
        // 声明交换机, 并指定类型
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 使用通道将消息发布到交换机 参数1: 指定交换机, 如果使用空字符串,表会使用默认的交换机
        channel.basicPublish(EXCHANGE_NAME, "error", null, MSG.getBytes());
        System.out.println("生产者发送消息 = " + MSG);
        // 关闭通道
        channel.close();
        // 关闭连接
        connection.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

消费者1

package com.rabbitmq.direct;

import com.rabbitmq.client.*;
import com.rabbitmq.utils.ConnectionUtil;

import java.io.IOException;

/**
 * direct 模式的消费者 1
 * Created by YongXin Xue on 2020/06/11 22:12
 */
public class Consumer_1 {
    private static final String RECV1_QUEUE_NAME = "recv1_direct_queue";
    // 交换机名称
    private static final String EXCHANGE_NAME = "direct_exchange";
    public static void main(String[] args) throws Exception {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列 : 生产者 -> 交换机 -> 队列 ->消费者
        channel.queueDeclare(RECV1_QUEUE_NAME,false, false, false, null);
        // 绑定队列到交换机
        // 参数1: 队列名称 参数2: 交换机名称 参数3: 接收打印的日志
        channel.queueBind(RECV1_QUEUE_NAME, EXCHANGE_NAME, "debug");
        channel.queueBind(RECV1_QUEUE_NAME, EXCHANGE_NAME, "info");
        channel.queueBind(RECV1_QUEUE_NAME, EXCHANGE_NAME, "warn");
        channel.queueBind(RECV1_QUEUE_NAME, EXCHANGE_NAME, "error");
        // 创建消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * 接收到数据的回调方法
             * @param consumerTag
             * @param envelope
             * @param properties
             * @param body  生产者发送过来的数据的字节数组
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("[direct 消费者 1 = " + msg + " ]");
            }
        };
        // 阻塞监听队列 true 设置自动对回执
        channel.basicConsume(RECV1_QUEUE_NAME, true, consumer);
    }
}

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

消费者2

package com.rabbitmq.direct;

import com.rabbitmq.client.*;
import com.rabbitmq.utils.ConnectionUtil;

import java.io.IOException;

/**
 * direct 模式的消费者 1
 * Created by YongXin Xue on 2020/06/11 22:12
 */
public class Consumer_2 {
    private static final String RECV2_QUEUE_NAME = "recv2_direct_queue";
    // 交换机名称
    private static final String EXCHANGE_NAME = "direct_exchange";
    public static void main(String[] args) throws Exception {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列 : 生产者 -> 交换机 -> 队列 ->消费者
        channel.queueDeclare(RECV2_QUEUE_NAME,false, false, false, null);
        // 绑定队列到交换机
        // 参数1: 队列名称 参数2: 交换机名称 参数3: 接收打印的error日志
        channel.queueBind(RECV2_QUEUE_NAME, EXCHANGE_NAME, "error");
        // 创建消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * 接收到数据的回调方法
             * @param consumerTag
             * @param envelope
             * @param properties
             * @param body  生产者发送过来的数据的字节数组
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("[direct 消费者 2 = " + msg + " ]");
            }
        };
        // 阻塞监听队列 true 设置自动对回执
        channel.basicConsume(RECV2_QUEUE_NAME, true, consumer);
    }
}

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

订阅模型-Topic

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

通配符规则：

`#`：匹配一个或多个词

`*`：匹配不多不少恰好1个词
1
2
3

举例：

`audit.#`：能够匹配`audit.irs.corporate` 或者 `audit.irs`

`audit.*`：只能匹配`audit.irs`
1
2
3

在这里插入图片描述

在这个例子中，我们将发送所有描述动物的消息。消息将使用由三个字（两个点）组成的routing key发送。路由关键字中的第一个单词将描述速度，第二个颜色和第三个种类：“<speed>.<color>.<species>”。
我们创建了三个绑定：Q1绑定了绑定键“.orange.”，Q2绑定了“..rabbit”和“lazy.＃”。

Q1匹配所有的橙色动物。
Q2匹配关于兔子以及懒惰动物的消息。

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】