【RabbitMQ】初识消息队列 MQ，基于 Docker 部署 RabbitMQ，探索 RabbitMQ 基本使用，了解常见的消息类型

文章目录

前言
一、初识消息队列 MQ
- 1.1 同步通信
- 1.2 异步通信
- 1.3 MQ 常见框架及其对比
二、初识 RabbitMQ
- 2.1 什么是 RabbitMQ
- 2.2 RabbitMQ 的结构
三、基于 Docker 部署 RabbitMQ
四、常见的消息类型
五、示例：在 Java 代码中通过 RabbitMQ 发送消息
- 5.1 消息发布者
- 5.2 消息消费者
- 5.3 使用 RabbitMQ 的原生 Java 客户端操作消息队列存的问题

前言

一、初识消息队列 MQ

1.1 同步通信

同步通信是指发起请求后，调用者需要等待服务提供者的响应。

同步通信的例子

使用手机打电话就是一种同步通信，此时我们只能与一个妹子进行通话，但是如果有另外的妹子想要和自己通话，那么就会建立通话失败，想想确实是一件遗憾的事情。

同步通信在程序中的调用问题

比如，微服务间基于 Feign 实现远程调用，而这种调用方式就是一种同步通信，例如下面微服务之间的调用关系图：

通过这个图示，可以很好的展示在微服务中，同步通信存在的弊端：

用户通过手机端调用了支付服务，再目前这个架构中，其他的服务如订单服务、仓储服务、短信服务等等在支付服务中的调用代码都是线性关系，也就是说只有当支付功能完成后还需要调用其他的服务，调用完所有的微服务之后，支付服务才算完成。
此时所有的微服务都是一个线性关系，因此用户支付所花费的时间就是所有微服务处理业务的时间总和了，此时带给用户的体验不佳不说，如果是面对高并发的场景，整个系统也很大可能会招架不住。
此时由于所有微服务都是线性关系的，如果系统中的一个微服务宕机了，那么整个系统就会崩溃。
如果要新增一个需求，即需要新增一个微服务，那么就会修改支付服务的代码，此时系统的耦合度较高。

因此，总体来说，同步调用存在如下问题：

耦合度高： 每次加入新的需求，都要修改原来的代码。
性能下降： 调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用的时间之和。
资源浪费： 调用链中的每个服务在等待响应过程中，不能释放请求占用的资源，高并发场景下会极度浪费系统资源。
级联失败： 如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌一样，迅速导致整个微服务群故障。

但是对于同步通信来说也具有优点：

例如，它的时效性很强，就如电话通信一样，可以实时获取对方的消息。

1.2 异步通信

异步通信是指发起请求后，调用者不需要立即等待服务提供者的响应。

异步通信的例子

例如，通过微信发送信息就是一种一步通信，当收到了消息之后，我们才去看收到的消息。并且同时可以向多个妹子发消息，一个不回就换另一个发，总有一个回消息的。

异步通信的方案——事件驱动模式

异步通信常见实现就是事件驱动模式，即某个消息就绪了，再通知其他服务来处理，如下图所示，微服务的调用就采用了事件驱动的模式：

通过这个图示，可以很好的展示在微服务中，异步通信的优势：

用户通过手机端调用支付服务，支付功能完成后将支付成功的消息发送给Broker，此时反馈给用户的耗时也就是这两步的耗时之后，此时的时间就非常短了，带给用户的体验也更佳。
这里的 Broker代表的是消息中间件。在消息传递中，Broker 是一种常见的模式。消息中间件充当了消息的中心处理单元，它接收来自生产者的消息，将其存储在队列中，并将消息传递给消费者。这种模式有助于实现解耦、异步通信和提高系统的可伸缩性。
当 Broker 收到了支付成功的消息之后，立即向其他的订阅者通知自己收到了支付成功的消息，然后其他的微服务再向 Broker 获取这个消息来完成自己的业务。
此时各种微服务之间的依赖性大大降低了，如果一个微服务宕机了并不会对整个系统造成太大的影响，并且如果想要新增微服务的话也不需要改动其他微服务的代码，降低了耦合度。
另外，Broker 也起到了消息缓存的作用，如果突然产生了大量的消息，可以缓存到 Broker中，而不至于对其他微服务造成过大的压力。

因此，总体来说，异步通信具体如下的优点：

耦合度低： 利用中间件 Broker 对各个微服务实现了解耦，即使新增业务也不会对其他微服务造成影响。
性能提升： 消息发布者只管将消息发送给中间件Broker，而无需关心其他微服务的执行。
故障隔离： 由于耦合度降低，因此一个微服务发生了故障，不会对其他微服务造成影响。
流量削峰： Broker 具有对消息的缓存作用，突然面对大量的并发的时候，可以起到缓冲作用。

但是异步通信也具有如下的缺点：

整个系统对消息中间件Broker 的可靠性依赖程度高，如果Broker 中间件异常了则会影响整个系统。
系统的架构更加复杂，业务没有明显的流程线，对业务管理追踪的难度大大提升。

1.3 MQ 常见框架及其对比

MQ 是 “消息队列” 的缩写，它是一种在分布式系统中用于进行异步通信的技术。消息队列允许不同的软件组件或系统之间通过在消息队列中发送和接收消息来进行通信，而不需要直接连接彼此，在上述的事件驱动架构中，Broker 就是 MQ。

在消息队列领域，有多种常见的框架，每个框架都有其优点和适用场景。一些常见的消息队列框架有：RabbitMQ、ActiveMQ、RocketMQ 和 Kafka 。它们都是消息中间件（Message Queue）系统，用于实现分布式系统中不同组件之间的异步通信。它们在设计和使用方面有一些区别，以下是它们的简要介绍：

RabbitMQ:
- 特点： RabbitMQ 是一个开源的消息代理软件，实现了高级消息队列协议（AMQP）。
- 设计理念： 设计简单、易于使用，支持广泛的消息传递模式，包括点对点、发布/订阅、请求/响应等。
- 语言支持： 支持多种编程语言，如Java、Python、Ruby等。
- 可靠性： 提供持久性消息、消息确认等机制，确保消息的可靠性传递。
ActiveMQ:
- 特点： ActiveMQ 是一个开源的消息中间件，实现了Java Message Service（JMS）规范。
- 设计理念： 面向Java应用，提供了丰富的API，支持多种消息传递模式和高级功能。
- 语言支持： 主要支持Java，但也有一些其他语言的客户端。
- 可靠性： 提供持久订阅、消息事务等功能，确保可靠性和一致性。
RocketMQ:
- 特点： RocketMQ 是阿里巴巴开源的分布式消息中间件系统。
- 设计理念： 设计为分布式、水平可扩展的系统，适用于大规模数据的处理。
- 语言支持： 提供Java、C++、Python等多语言的客户端。
- 可靠性： 支持多种消息传递模式，具备高可用性、高吞吐量和低延迟的特性。
Kafka:
- 特点： Kafka 是由Apache软件基金会开发的分布式流处理平台，兼具消息队列和日志系统的功能。
- 设计理念： 设计为高吞吐、持久性、可水平扩展的分布式系统，用于处理实时数据流。
- 语言支持： 提供Java和Scala等语言的客户端。
- 可靠性： 提供副本机制、持久性日志存储等特性，适用于大规模数据流的处理。

如下表所示，总结了这四种消息队列在不同方面的差异：

特性	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议支持	AMQP, XMPP, SMTP, STOMP	OpenWire, STOMP, REST, XMPP, AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般	差	高	非常高
消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

这个表格展示了这四个消息队列框架的一些关键特性的对比。总体而言，选择其中一个消息中间件系统通常取决于具体的业务需求、系统架构以及性能要求。不同的系统可能更适合不同的消息中间件。

二、初识 RabbitMQ

2.1 什么是 RabbitMQ

RabbitMQ 是一个开源的消息代理软件，实现了高级消息队列协议（AMQP）。它允许不同应用之间进行异步通信，并提供了一种有效的方式来处理分布式系统中的大量消息。

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

下面是 RabbitMQ 的一些关键特点：

消息代理： RabbitMQ 充当消息代理，负责接收、存储和转发消息。
消息队列： 通过消息队列实现消息的存储和传递，允许生产者将消息发送到队列，而消费者从队列中接收消息。
消息模型： 支持多种消息传递模型，包括点对点、发布/订阅、请求/响应等。
可靠性： 提供持久性消息、消息确认等机制，确保消息的可靠传递。
灵活性： RabbitMQ 是高度可配置的，支持多种插件和扩展，可以适应不同的应用场景。
跨平台： 支持多种操作系统，并提供多种语言的客户端，如 Java、Python、Ruby 等。

RabbitMQ 的设计理念是简单、灵活、可靠，并且具有广泛的应用领域，从企业级应用到小型项目都能发挥其优势。通过使用 RabbitMQ，开发人员可以更轻松地实现分布式系统中的消息传递和解耦。

2.2 RabbitMQ 的结构

RabbitMQ 的整体结构如下图所示：

RabbitMQ Structure

在上图中，我们可以看到 RabbitMQ 的结构主要涉及以下几个核心概念：

Producer（生产者）： 生产者是消息的发送方，负责将消息发送到 RabbitMQ 服务器。
Exchange（交换机）： 交换机接收来自生产者的消息，并将其路由到一个或多个队列。交换机有不同的类型，包括直连交换机（direct）、主题交换机（topic）、扇出交换机（fanout）等，用于定义消息的路由规则。
Queue（队列）： 队列是消息的存储位置，生产者或交换机将消息发送到队列，而消费者则从队列中接收消息进行处理。
Binding（绑定）： 绑定定义了 Exchange 和 Queue 之间的关系，指定消息如何从 Exchange 路由到特定的队列。绑定规则根据交换机的类型而异。
Consumer（消费者）： 消费者是消息的接收方，负责从队列中获取消息并进行相应的处理。
VirtualHost（虚拟主机）： 虚拟主机是逻辑上的隔离单位，每个虚拟主机都拥有自己的独立的用户、交换机、队列等资源，用于隔离不同应用或团队的消息流。

RabbitMQ 的结构允许构建灵活的消息传递系统，支持多种消息传递模式和路由策略，使其适用于各种不同的应用场景。这种灵活性使得 RabbitMQ 成为分布式系统中常用的消息中间件。

三、基于 Docker 部署 RabbitMQ

RabbitMQ的部署分为单机部署和集群部署，这里以单机部署为例。在 Centos7 虚拟机中使用 Docker 部署 RabbitMQ，以下是单例部署的步骤：

拉取 RabbitMQ 镜像

使用以下命令拉取带有管理插件的 RabbitMQ 镜像：
```
docker pull rabbitmq:3-management
```

启动 RabbitMQ 容器

执行以下命令启动 RabbitMQ 容器：

docker run \
-e RABBITMQ_DEFAULT_USER=yourname \
-e RABBITMQ_DEFAULT_PASS=yourpassword \
--name mq \
--hostname mq1 \
-p 15672:15672 \
-p 5672:5672 \
-d \
rabbitmq:3-management

yourname：自定义的 RabbitMQ 用户名。
yourpassword：自定义的 RabbitMQ 密码。

例如：

访问 RabbitMQ 管理界面

启动成功后，通过浏览器访问 宿主机IP:15672，使用上述用户名和密码登录 RabbitMQ 管理界面。

在管理界面中，我们可以监控队列、交换机，查看连接、通道等信息。

通过上述步骤，就完成了基于 Docker 的 RabbitMQ 单例部署。在实际生产环境中，可能还需要考虑持久化配置、集群部署等问题。

四、常见的消息类型

RabbitMQ 提供了多种消息传递模型，常见的消息类型包括以下几种，对应 RabbitMQ 官方文档中的不同示例：

基本消息队列（BasicQueue）
- 描述： 这是最简单的消息队列模型，一个生产者发送消息到队列，一个消费者从队列中接收并处理消息。
工作消息队列（WorkQueue）
- 描述： 多个消费者共享一个队列，生产者发送消息到队列，其中一个消费者接收并处理消息。适用于任务分发和负载均衡。
发布订阅（Publish、Subscribe）
- 广播（Fanout Exchange）
  - 描述： 生产者将消息发送到交换机，交换机将消息广播到所有与之绑定的队列。每个队列都有自己的消费者。
- 路由（Direct Exchange）
  - 描述： 生产者发送带有指定路由键的消息到交换机，交换机根据路由键将消息路由到匹配的队列。每个队列有一个或多个消费者。
- 主题（Topic Exchange）
  - 描述： 类似于直连交换机，但是主题交换机允许使用通配符进行匹配，以实现更灵活的消息路由。

这些不同的消息类型满足了不同的应用场景需求，使 RabbitMQ 成为一个灵活而强大的消息中间件。通过选择合适的消息模型，开发人员可以更好地满足系统设计的要求。

五、示例：在 Java 代码中通过 RabbitMQ 发送消息

5.1 消息发布者

以下是一个通过 Java 中单元测试的代码向 RabbitMQ 发送消息的示例。在这个示例中，我们使用 RabbitMQ 的 Java 客户端库来创建连接、通道，并发送消息到队列。

public class PublisherTest {@Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.146.128");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("admin");factory.setPassword("123456");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.发送消息String message = "Hello, RabbitMQ!";channel.basicPublish("", queueName, null, message.getBytes());System.out.println("发送消息成功：【" + message + "】");// 5.关闭通道和连接channel.close();connection.close();}
}

对上述代码的说明：

建立连接： 使用 ConnectionFactory 创建连接，并设置 RabbitMQ 服务器的地址、端口号、用户名和密码。
创建通道： 通过连接创建通道，大部分的 RabbitMQ 操作都是通过通道进行的。
创建队列： 使用通道创建一个名为 simple.queue 的队列，如果该队列不存在则会被创建。
发送消息： 使用 basicPublish 方法发送消息到指定的队列。在本例中，消息内容为 “Hello, RabbitMQ!”。
关闭通道和连接： 释放资源，关闭通道和连接。

然后运行这段代码：

可以发现在 RabbitMQ 的管理页面就多了一个 simple.queue 的队列，并且其中有一条未消费的消息：

点击消息的名称，然后在点击获取消息，就可以看到具体的消息内容了。

5.2 消息消费者

以下是一个通过 Java中单元测试的代码实现的 RabbitMQ 消息消费者示例。在这个示例中，我们使用 RabbitMQ 的 Java 客户端库来创建连接、通道，并从队列中接收和处理消息。

public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.146.128");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("admin");factory.setPassword("123456");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息channel.basicConsume(queueName, true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {// 5.处理消息String message = new String(body);System.out.println("接收到消息：【" + message + "】");}});System.out.println("等待接收消息。。。。");}
}

对上述代码的说明：

建立连接： 使用 ConnectionFactory 创建连接，并设置 RabbitMQ 服务器的地址、端口号、用户名和密码。
创建通道： 通过连接创建通道，大部分的 RabbitMQ 操作都是通过通道进行的。
创建队列： 使用通道创建一个名为 simple.queue 的队列，如果该队列不存在则会被创建。
订阅消息： 使用 basicConsume 方法订阅消息。在本例中，我们使用匿名内部类继承 DefaultConsumer 类，并重写 handleDelivery 方法来处理接收到的消息。
处理消息： 在 handleDelivery 方法中，处理接收到的消息。在本例中，我们简单地打印出接收到的消息。