RabbitMQ（一）——入门

2023-10-29

前言

原来公司项目的消息中间件一直在用RabbitMQ,今天抽出时间简单总结梳理一下关于RabbitMQ的相关知识点。我们知道消息队列在分布式系统中应用的的地方有很多，它也有很多种类型，除了今天重点介绍的RabbitMQ，还有像ActiveMQ，阿里的RocketMQ，在大数据场景中经常用到的Kafka，还有一些其他的诸如ZeroMQ，MetaMQ等等。ActiveMQ在之前其实一直很火，许多公司的项目也都在使用，但近来它的社区变得不太活跃，因此工作中遇到相关问题的时候，对应的解决方案不是太好找，所以现在ActiveMQ慢慢变得不如从前那么流行了，而RabbitMQ、Kafka等正在慢慢的占据主流。

不废话了，开始讲解我们今天的重点RabbitMQ吧。今天入门主要讲解下面三个部分：

一、RabbitMQ的下载安装

二、RabbitMQ的工作原理

三、RabbitMQ入门测试

一、RabbitMQ的下载安装

RabbitMQ由Erlang语言开发的，所以我们要向在项目中用RabbitMQ，除了要安装RabbitMQ外还需要安装erlong语言以确保对RabbitMQ的环境支持。另外需要注意的一点是在安装RabbitMQ和erlong的时候，要确保二者的版本相互匹配兼容，我用的RabbitMQ是3.7.10版本erlong是21.3版本。

1.下载erlong

erlong下载地址：http://www.erlang.org/downloads

下载好以后，以管理员方式运行此文件，安装，除了安装目录可以自己选择一下外，其他的一路下一步即可。

同java一样，erlang安装完成需要配置erlang的环境变量：

ERLANG_HOME=D:\others\erlong\erl10.3

在path中添加%ERLANG_HOME%\bin

2.下载RabbitMQ

RabbitMQ的下载地址：http://www.rabbitmq.com/download.html

下载完成后，以管理员方式运行此文件，安装。

3.启动RabbitMQ

安装成功后会自动创建RabbitMQ服务并且启动。我们可以到系统服务列表中找一下，如下图已经启动了

如果没有启动则进入安装目录下的sbin目录手动启动：

4.安装RabbitMQ的管理插件

安装rabbitMQ的管理插件，主要是方便我们在浏览器端输入网址后进入RabbitMQ的后台管理RabbitMQ。

在命令行下进入RabbitMQ安装目录的sbin目录下，输入指令：rabbitmq-plugins.bat enable rabbitmq_management

之后我们就可以通过http://localhost:15672进入到RabbitMQ的管理后台了，如下（账号密码默认的均为guest）：

二、RabbitMQ的工作原理

上图是RabbitMQ的基本结构，我们来对其中的各个部分做简单说明：

Producer ：消息生产者，即生产方客户端，生产方客户端将消息发送到MQ。
Consumer ：消息消费者，即消费方客户端，接收MQ转发的消息。
Broker ：消息队列服务进程，此进程包括两个部分：Exchange（交换机）和Queue（队列）。
Exchange ：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。

首先消息生产者想要发送消息给RabbitMQ必须先与其建立连接（connect）这个连接的底层是通过socket实现的，我们不必太过关心。在连接的内部其实有很多通道（Channel），具体的消息发送其实就是通过这些通道发送到RabbitMQ的。而消息的消费者一方要想拿到RabbitMQ中的消息也必须与其建立连接，这个连接（包括里面的通道）和生产者一方是一样的。对上方的几个概念有大体的了解即可，我们会在接下来的部分结合实际案例来详细说明各个部分的功能作用。

总结——消息发布接收流程：

发送消息：

生产者和Broker建立TCP连接
生产者和Broker建立通道
生产者通过通道消息发送给Broker，由Exchange将消息进行转发
Exchange将消息转发到指定的Queue（队列）

接收消息：

消费者和Broker建立TCP连接
消费者和Broker建立通道
消费者监听指定的Queue（队列）
当有消息到达Queue时Broker默认将消息推送给消费者
消费者接收到消息

三、RabbitMQ入门测试

进过第一步的RabbitMQ安装和第二步RabbitMQ原理的简单介绍，现在我们简单的写个测试，让消息生产端发送一个“hello world”到MQ,然后由MQ转到消息消费端。这里需要注意一点：不管是消息的生产者和还是消费者，相对于RabbitMQ它们二者都属于客户端。所以在创建对应的对应时要添加RabbitMQ的客户端依赖，我们这里先用 rabbitMQ官方提供的java client测试，目的是对RabbitMQ的交互过程有个清晰的认识。

1.创建maven工程

创建生产者工程和消费者工程，分别加入RabbitMQ客户端 java client的依赖。

test-rabbitmq-producer：生产者工程
test-rabbitmq-consumer：消费者工程

二者的pom.xml文件相同，都为如下内容

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
        http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <!--每一个springboot工程都必须有这个依赖-->
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.1.RELEASE</version>
    </parent>

    <modelVersion>4.0.0</modelVersion>
    <artifactId>test-rabbitmq-producer</artifactId>

    <dependencies>
        <!--RabbitMQ的客户端依赖-->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.7.0</version>
        </dependency>
        <!--记录日志-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-logging</artifactId>
        </dependency>
    </dependencies>
</project>

2.生产者Producer01

在生产者工程下的test中创建测试类如下：

这里是作为入门程序为了让大家对RabbitMQ的工作过程有大体了解，并对每个参数大体有个印象，所以代码有点多。其实在实际项目中通过在springboot中引入别的客户端依赖，我们只需要写很少的代码就可以完成下面同样的功能。

public class Producer01 {
    //申明队列
    private static final String QUEUE = "helloworld";
    public static void main(String[] args) {
        //第一步：生产者和Broker建立TCP连接。
        //通过连接工厂创建新的连接和mq建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置ip地址，我们在本地所以直接是127.0.0.1
        connectionFactory.setHost("127.0.0.1");
        //设置端口，RabbitMQ默认端口即为5672
        connectionFactory.setPort(5672);
        //设置账号和密码
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq。这里不理解可以暂时略过
        connectionFactory.setVirtualHost("/");

        Connection connection = null;
        Channel channel = null;
        try {
            //建立新连接
            connection = connectionFactory.newConnection();
            //第二步：生产者和Broker建立通道。
            //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
            channel = connection.createChannel();
            //声明队列，如果队列在mq中已经存在则使用已有的，如果没有则创建新的
            /**
             * 参数明细：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             * 1、queue 队列名称
             * 2、durable 是否持久化，如果持久化，mq重启后队列还在
             * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
             * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
             * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间等
             */
            channel.queueDeclare(QUEUE,true,false,false,null);
            //第三步：生产者通过通道消息发送给Broker，由Exchange将消息进行转发。
            /**
             * 参数明细：String exchange, String routingKey, BasicProperties props, byte[] body
             * 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）
             * 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称
             * 3、props，消息的属性
             * 4、body，消息内容
             */
            //消息内容
            String message = "hello world!";
            channel.basicPublish("",QUEUE,null,message.getBytes());
            System.out.println("send to mq "+message);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //关闭连接
            //先关闭通道
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                //再关闭连接
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

3.生产者Consumer01

在消费者工程下的test中创建测试类如下：

public class Consumer01 {
    //申明队列，要与生产者的保持一致
    private static final String QUEUE = "helloworld";
    public static void main(String[] args) throws IOException, TimeoutException {
        //第一步：消费者和Broker建立TCP连接
        //通过连接工厂创建新的连接和mq建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
        connectionFactory.setVirtualHost("/");

        //建立新连接
        Connection connection = connectionFactory.newConnection();
        //第二步：消费者和Broker建立通道。
        //创建会话通道,消费者和mq服务所有通信都在channel通道中完成
        Channel channel = connection.createChannel();

        //第三步：消费者监听指定的Queue（队列）
        //声明队列，如果队列在mq中已经存在则使用已有的，如果没有则创建新的
        /**
         * 参数明细：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
         * 1、queue 队列名称
         * 2、durable 是否持久化，如果持久化，mq重启后队列还在
         * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
         * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
         * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
         */
        channel.queueDeclare(QUEUE,true,false,false,null);
        //第四步：实现消费方法，当有消息到达Queue时Broker默认将消息推送给消费者。
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            /**
             * 当接收到消息后此方法将被调用，
             * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
             * @param envelope 信封，通过envelope
             * @param properties 消息属性
             * @param body 消息内容
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //交换机
                String exchange = envelope.getExchange();
                //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
                long deliveryTag = envelope.getDeliveryTag();
                //消息内容
                String message= new String(body,"utf-8");
                System.out.println("receive message:"+message);
            }
        };
        //监听队列
        /**
         * 参数明细：String queue, boolean autoAck, Consumer callback
         * 1、queue 队列名称
         * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
         * 3、callback，消费方法，当消费者接收到消息要执行的方法
         */
        channel.basicConsume(QUEUE,true,defaultConsumer);
    }
}

4.运行测试

现在我们先把Producer01类运行起来，然后通过浏览器登录到RabbitMQ的后台，可以看到总共有一条待发送的消息，如下：

这时我们再把Consumer01这个类启动起来，在控制台看到如下日志，说明消费端启动起来以后，mq就把其队列中从producer01接收到的消息发送给了Consumer01。再到RabbitMQ的后台看一看发现现在的待发送消息已经变为了0条

receive message:hello world!

5.流程总结

通过上面的代码编写，我们简单的总结一下生产端和消费端的操作流程。

生产端：

创建连接
创建通道
声明队列
发送消息

消费端：

创建连接
创建通道
声明队列
监听队列
接收消息
ack回复

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

消息中间件

rabbitmq