揭开消息中间件RabbitMQ的神秘面纱

Posted 2020-11-19 xingyunblog

当你看到这篇博文的时候，相信你至少已经知道RabbitMQ 是一个非常优秀的消息中间件，它使用专门处理高并发的Erlang 语言编写而成的消息中间件产品。

本文我们将重点学习消息队列，消息中间件的概念，以及如何在Windows 上安装RabbitMQ并使用它发送一个消息

为了更好地学习RabbitMQ还是先来看看一些专业平台是如何解释它的吧。

1. 关于RabbitMQ的解释

RabbitMQ是部署最广泛的开源消息代理。

RabbitMQ在全球范围内在小型初创公司和大型企业进行了超过35,000次的生产部署，是最受欢迎的开源消息代理。

RabbitMQ轻巧且易于部署在云端和云端。 它支持多种消息传递协议。 RabbitMQ可以部署在分布式和联合配置中，以满足高规模，高可用性需求。

RabbitMQ可运行在许多操作系统和云环境中，并为大多数流行语言提供广泛的开发工具。

————官网解释

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。

RabbitMQ服务器是用Erlang语言编写的，而群集和故障转移是构建在开放电信平台框架上的。

———— 维基百科

Tips: 

在最近研读的Think in Java 第四版中有这样一段话：

如果你发现程序中某个部分必须大量使用并发，并且你在试图构建这个部分时碰到了过多的问题，那么你可以考虑使用Erlang 这类专门的并发语言来创建这个部分。

1.1 什么是消息队列？

1.2 那什么是消息中间件呢？

消息队列中间件（简称消息中间件）是指利用高效可靠的消息传递机制进行与平台无关的数据交流，并基于数据通信来进行分布式系统的集成。

通过提供消息传递和消息排队模型，它可以在分布式环境下提供应用解耦、弹性伸缩、冗余存储、流量削峰、异步通信、数据同步等等功能，

其作为分布式系统架构中的一个重要组件，有着举足轻重的地位.

传统的信息系统作法可能是收到一条消息就马上同步存入数据库，这种作法在小并发量的情况下可以很好的工作，但互联网大并发环境下就是灾难.

更多解释请参考：消息队列之 RabbitMQ

---

 2. 消息中间件知多少？

ActiveMQ 是 Apache 出品的、采用 Java 语言编写的完全基于 JMS1.1 规范的面向消息的中间件，为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。不过由于历史原因包袱太重，目前市场份额没有后面三种消息中间件多，其最新架构被命名为 Apollo，号称下一代 ActiveMQ，有兴趣的同学可行了解。

RabbitMQ 是采用 Erlang 语言实现的 AMQP 协议的消息中间件，最初起源于金融系统，用于在分布式系统中存储转发消息。RabbitMQ 发展到今天，被越来越多的人认可，这和它在可靠性、可用性、扩展性、功能丰富等方面的卓越表现是分不开的。

Kafka 起初是由 LinkedIn 公司采用 Scala 语言开发的一个分布式、多分区、多副本且基于 zookeeper 协调的分布式消息系统，现已捐献给 Apache 基金会。它是一种高吞吐量的分布式发布订阅消息系统，以可水平扩展和高吞吐率而被广泛使用。目前越来越多的开源分布式处理系统如 Cloudera、Apache Storm、Spark、Flink 等都支持与 Kafka 集成。

RocketMQ 是阿里开源的消息中间件，目前已经捐献个 Apache 基金会，它是由 Java 语言开发的，具备高吞吐量、高可用性、适合大规模分布式系统应用等特点，经历过双 11 的洗礼，实力不容小觑。

ZeroMQ 号称史上最快的消息队列，基于 C 语言开发。ZeroMQ 是一个消息处理队列库，可在多线程、多内核和主机之间弹性伸缩，虽然大多数时候我们习惯将其归入消息队列家族之中，但是其和前面的几款有着本质的区别，ZeroMQ 本身就不是一个消息队列服务器，更像是一组底层网络通讯库，对原有的 Socket API 上加上一层封装而已。

目前市面上的消息中间件还有很多，比如腾讯系的 PhxQueue、CMQ、CKafka 等，甚至NoSQL Redis据说 也有消息 队列的功能。

参考博文：IM系统的MQ消息中间件选型：Kafka还是RabbitMQ？

3.RabbitMQ安装

RabbitMQ是一个AMQP（Advanced Message Queue，即高级消息队列协议）服务器 。

下载地址： RabbitMQ下载

安装说明：各平台下RabbitMQ安装指南

3.1 Windows 下安装RabbitMQ

1. 点击 RabbitMQ下载 我们可以看到这个页面

2. 这里我们选择图中推荐的  Windows 安装RabbitMQ 文档

 下载完成后选中*.exe 文件右键管理员身份运行，一般情况下会出现这个

Tips: 这是怎么回事呢?

由于RabbitMQ 是由Erlang语言编写而成，所以一般情况在安装RabbitMQ之前，我们还需要安装Erlang的运行环境，类似java中的JRE或者C#中的 .net framework.

 3. 下载安装Erlang

点击是，会自动打开Erlang 的官方下载页面

根据自己的电脑选择对应的版本，我的电脑是windows 10 64 位，所以这里选择64 位，如果你的电脑是32 位清选择32 位版本。

这里默认是没有勾选的，我们需要勾选下，然后点击 Next，根据提示安装完成即可。

安装的时候默认我发现已经生成了

Tips: 如果没有请自行添加

将Erlang 添加到Path路径下

%ERLANG_HOME%in

 

4. 安装RabbitMQ

安装完成Erlang 之后，我们再次运行我们的RabbitMQ 安装包

默认的安装路径是C:Program FilesRabbitMQ Server abbitmq_server-3.7.6

5. 配置环境变量

一般情况下，我们最好配置下环境变量以便于我们今后更好地使用。

RABBITMQ_BASE

C:Program FilesRabbitMQ Server
abbitmq_server-3.7.6

 然后我们需要添加到Path变量中

%RABBITMQ_BASE%sbin

 6. 重新安装service

要使环境更改在Windows上生效，必须重新安装该服务。 重启服务是不够的。

这可以使用安装程序或者在具有管理员权限的命令行上完成。

注意：有两个文件rabbitmq-service.bat 和 rabbitmq-server.bat 别弄混了，不然执行命令会失败

管理员权限运行cmd 命令行，进入安装文件夹下的sbin目录

这个RabbitMQ service 服务是自动开启的，所以我们需要先停止RabbitMQ服务

rabbitmq-service.bat stop

然后移除RabbitMQ服务

rabbitmq-service.bat remove

再次安装

rabbitmq-service.bat install

 

总结：关于RabbitMQ Service的用法

7. 检测RabbitMQ 运行状态

如果我们想查看RabbitMQ 的运行状态，那么输入下列命令即可

rabbitmqctl status

 但是当年你输入后可能会看到这样的错误信息

解决方案：

C:WindowsSystem32configsystemprofile.erlang.cookie

C:Usersfairy.erlang.cookie

对比下这两个文件，你会发现里面的值不一样，这就是导致这个错误的原因。

我们只需要将其中一个替换掉，两个保持统一即可，比如将系统下的那个文件替换掉个人用户下的那个文件

再次执行命令

rabbitmqctl status

执行成功后可以看到下面的回显：

8.安装 RabbitMQWeb的管理插件

执行命令

rabbitmq-plugins enable rabbitmq_management

要 生效需要重启下RabbitMQ Server

9. 查看当前用户列表

rabbitmqctl.bat list_users

 执行成功后可以看到如下回显：

10.打开RabbitMQ Web 管理界面

http://127.0.0.1:15672/

 执行成功后可以看到这个界面

默认端口是15672

账号和密码默认都是guest

登陆后可以看到这样的管理界面

4. 打造第一个Hello World 程序

RabbitMQ是一个消息代理：它接受和转发消息。你可以把它想象成一个邮局：当你把你想要发布的邮件放在邮箱里时，你可以确定先生或女士邮递员最终将邮件发送给你的收件人。

在这个比喻中，RabbitMQ是邮政信箱，邮局和邮递员。

RabbitMQ和邮局的主要区别在于它不处理纸张，而是接受，存储和转发二进制数据块 - 消息。

英文原文地址：https://www.rabbitmq.com/tutorials/tutorial-one-java.html

在搭建这个Hello World 之前，我们有必要了解下 RabbitMQ和一般的消息传递使用的一些术语。

4.1  消息传递术语

4.1.1 生产者

制作意味着发送。 一个发送消息的程序是一个生产者：

Tips: P 代表生产者

4.1.2 队列

• 队列是生活在RabbitMQ中的邮箱的名称。 尽管消息流经RabbitMQ和您的应用程序，但它们只能存储在队列中。
• 一个队列只受主机内存和磁盘限制的约束，它本质上是一个很大的消息缓冲区。
• 许多生产者可以发送进入一个队列的消息，并且许多消费者可以尝试从一个队列接收数据。 这就是我们代表队列的方式：

 

4.1.3 消费者

消费与接受有类似的意义。 消费者是一个主要等待接收消息的程序：

请注意，生产者，消费者和中间件不必驻留在同一主机上;

4.2 "Hello World"使用 Java 客户端

在本教程的这一部分中，我们将用Java编写两个程序; 一个发送单个消息的生产者，以及接收消息并将其打印出来的消费者。

我们将详细介绍Java API中的一些细节，专注于这个非常简单的事情，以便开始使用。 这是一个消息传递的“Hello World”。

在下图中，“P”是我们的生产者，“C”是我们的消费者。 中间的盒子是一个队列 - RabbitMQ代表消费者保存的消息缓冲区。

 

RabbitMQ 有许多不同语言的RabbitMQ客户端。 这里我们将使用RabbitMQ提供的Java客户端。

4.2.1  下载相关依赖Jar 包

下载RabbitMQ提供的Java客户端以及它的依赖(SLF4J API and SLF4J Simple)

将这些文件复制到工作目录中，并跟着教程复制Java文件。

Tips:

RabbitMQ Java客户端也位于中央Maven存储库中，其中包含groupId com.rabbitmq和artifactId amqp-client。

请注意SLF4J Simple对于教程来说已经足够了，但是您应该在生产环境中使用像Logback 这样的完整日志库。

 关于Maven 等其他下载方式请移步

Java Client

• On Maven Central: RabbitMQ Java client.
• Quick download: Binary | Source
• Javadoc
• All Java client downloads
• Older versions

4.2.2 编写代码

我们拥有Java客户端及其依赖关系，那么我们接下来开始写代码。

由于是学习使用RabbitMQ,我们这里使用 STS 来写这个项目。

1. 打开STS,新建一个名字叫做 RabbitMQ_HelloWorld_Sample  的 Java Project。

2. 新建一个叫做libs的文件夹，我们将上面三个jar 复制到我们的项目中，然后并添加依赖

 

 完成后项目结构如图所示

3. 编写生产者类

我们会打电话给我们的消息发布者（发送者）发送和我们的消息使用者（接收者）Recv。 发布者将连接到RabbitMQ，发送一条消息，然后退出

创建Send.java 文件，关于代码的讲解在注释里：

import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Send {

    //设置消息队列的名称
    private final static String QUEUE_NAME="hello";
    
    public static void main(String[] args) throws java.io.IOException, TimeoutException{
        
        /**
         * 创建一个到RabbitMQ Server 的连接
         * 连接抽象出套接字连接，并为我们处理协议版本协商和身份验证等。 
         * 在这里，我们连接到本地机器上的代理 - 因此是本地主机。 
         * 如果我们想连接到另一台机器上的代理，我们只需在此指定其名称或IP地址。
         * */
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        
        /**
         * 接下来我们创建一个Channel 对象，这是大部分用于完成任务的API驻留的地方。
         * 要想发送出去，我们必须声明一个队列来执行发送,那么我们可以将消息发布到队列中：
         * */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String message = "Hello World!";
        //声明一个队列是幂等的 - 只有当它不存在时才会被创建。 
        //消息内容是一个字节数组，所以你可以编码任何你喜欢的地方。
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent ‘" + message + "‘");
        
        //最后我们关闭这些连接对象
        channel.close();
        connection.close();
        
    }
} 

 执行成功后你将看到这样的内容：

Tips: ，

如果这是您第一次使用RabbitMQ，并且您没有看到“已发送”消息，那么您可能会抓住您的头脑，想知道会出现什么问题？

也许代理启动时没有足够的可用磁盘空间（默认情况下它至少需要200 MB空闲空间），因此拒绝接受消息。

检查代理日志文件以确认并在必要时减少限制。 配置文件文档将告诉你如何设置disk_free_limit。

还有一种可能是你的RabbitMQ 没有启动，执行下面命令再次尝试即可。

rabbitmq-service.bat start

 4. 编写消费者

这就是我们的出版商。 我们的消费者推送来自RabbitMQ的消息，因此与发布单个消息的发布者不同，我们将继续运行以收听消息并将其打印出来。

创建一个Recv.java 文件，代码讲解在注释里面

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

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 消费者
 * 额外的DefaultConsumer是一个实现Consumer接口的类，我们将使用它来缓存由服务器推送给我们的消息。
 * */
public class Recv {
     private final static  String QUEUE_NAME="hello";
     
     public static void main(String[] args) throws IOException, TimeoutException {
         
          /**
           * 设置与发布者相同; 
           * 我们打开一个连接和一个通道，并声明我们将要使用的队列。 
           * 请注意，这与发送发布到的队列相匹配
           * */
           ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            /**
             * 请注意，我们也在这里声明队列。 
             * 因为我们可能会在发布者之前启动消费者，所以我们希望在我们尝试使用消息之前确保队列已存在。   
             * */
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
            
            /**
             *  我们即将告诉服务器将队列中的消息传递给我们。
             *   由于它会异步推送消息，因此我们以对象的形式提供回调，该消息将缓冲消息，直到我们准备好使用它们。 
             *  这是一个DefaultConsumer子类的作用。
             * */

            Consumer consumer = new DefaultConsumer(channel) {
                  @Override
                  public void handleDelivery(String consumerTag, Envelope envelope,
                                             AMQP.BasicProperties properties, byte[] body)
                      throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println(" [x] Received ‘" + message + "‘");
                  }
                };
            channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

执行成功后

我们可以看到我们的消费者收到了消息队列生产者刚发布的消息。

本篇完~