[大数据] Windows 下安装RabbitMQ服务器及基本配置

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Windows 下安装RabbitMQ服务器及基本配置 -> 正文阅读

[大数据]Windows 下安装RabbitMQ服务器及基本配置

1.安装Erlang

安装Erlang 时要注意安装的RabbityMQ 所依赖的Erlang版本，根据RabbitMQ的要求选择一个版本，这里安装的RabbitMQ的版本是 3.8.19 ，他依赖的Erlang版本范围是23.2到 24.x，因此我选择版本是OTP 24.0.Erlang 下载地址。
在这里插入图片描述

查看RabbitMQ所依赖的Erlang版本：https://www.rabbitmq.com/which-erlang.html

下载Erlang安装包后，直接双击安装就可以了。
双击，如下：
在这里插入图片描述
点击next,如下：

更换路径，点击next,如下：

点击Install,进行安装，稍等会，安装成功：

2.设置ERLANG_HOME 环境变量

在这里插入图片描述
变量名：ERLANG_HOME
变量值: erlang的安装地址
点击确定在系统环境变量里就增加刚刚添加的ERLANG_HOME

然后，双击系统变量Path
点击新建，将%ERLANG_HOME%\bin加入到Path中。

最后windows键+R键，输入cmd,再输入erl,看到版本号就说明erlang安装成功了。
在这里插入图片描述

注意：如果之前安装了Erlang的其他版本，需要卸载后在进行重新安装和设置。

接下来安装RabbitMQ。

3.安装RabbitMQ

可以在RabbitMQ的官方网站下载最新版本的RabbitMQ服务器安装程序。
RabbitMQ下载地址，这里我下载的是官方推荐的最新版本rabbitmq-server-3.8.19.exe，然后点击默认安装。

在这里插入图片描述

RabbitMQ安装好后是作为windows service 运行在后台。
在这里插入图片描述

4.设置RabbitMQ环境变量

RabbitMQ 安装好后在开始菜单输入rabbitmq 会看到如下几个菜单：
在这里插入图片描述
红色圈内的三个菜单是提供控制Windows service的命令，为了能够在任意Windows命令窗口上操控RabbitMQ服务需要在系统里加一个环境变量并且配置在系统的PHTH环境变量中。

首先添加一个RABBITMQ_SERVER变量：

在这里插入图片描述
然后在系统的path变量中如下配置：

这样就可以在windows administrator启动的CMD窗口操控RabbitMQ服务了。
不需要每次都定位到：
D:\Program Files\RabbitMQ Server\rabbitmq_server-3.8.19\sbin>

5.安装rabbitmq-management

我们可以使用命令查看一下RabbitMQ的所有插件：

C:\Users\qzz>rabbitmq-plugins list

看到RabbitMQ罗列出来了很多插件
在这里插入图片描述
我们用下列命令安装rabbitmq_management插件，这款插件是可视化的方式查看RabbitMQ服务器实例化的状态，以及操控RabbitMQ服务器。

C:\Users\qzz>rabbitmq-plugins enable rabbitmq_management

运行完命令后，看到如下图表示安装成功了。
在这里插入图片描述
现在我们在浏览器中输入：http://localhost:15672可以看到一个登录界面：
这里可以使用默认账号guest/guest登陆后的界面如下：

在浏览器中输入http://localhost:15672/api/index.html就可以看到Rabbit MQ Management HTTP API文档，如下图：

在这里插入图片描述
这样就可以查看Rabbit MQ server实例状态的相关信息了。

6.管理rabbitmq_management的用户

用命令rabbitmqctl list_users 查看一下现rabbitmq_management注册用户：
在这里插入图片描述
发现现在只有一个用户guest，并且它的tag是administrator。

那么在命令行下创建一个用户，创建用户的命令是:

rabbitmqctl add_user [username] [password]

现在创建一个username=rabbit1 password=rabbit1的用户，命令如下：

rabbitmqctl add_user rabbit1 rabbit1

创建成功了：
在这里插入图片描述
现在再看下有多少用户，执行一下命令：

rabbitmqctl list_users

在这里插入图片描述
发现用户多了一个rabbit1，但是tag是空的。使用命令给用户rabbit1设置tag,
设置tag的命令是：

rabbitmqctl set_user_tags [tag1] [tag2] ...

一次可以给一个用户设置多个tag，也可以设置一个
在这里插入图片描述
现在用户rabbit1有两个tag了，一个是administrator，1个是none。

有5个tag可供选择:
administrator、monitoring、policymaker、management和none。

其实这里的tag代表的是权限，administrator是最高权限，none表示不能访问，这里administrator和none的组合，权限应该是向高看齐，忽略none，用的是administrator的权限。

有兴趣的同学可以到这里了解各个tag的含义

tag含义如下：
在这里插入图片描述
我们用rabbit1/rabbit1 登录rabbitmq_management。

其实有了rabbitmq_management这个可视化插件，很多事情都可以在这个插件里干，包括创建用户，创建交换机（Exchange)和创建队列（Queque）。

关于Rabbit的在windows下的基本配置就告一段落，关于更多的更高级的配置，可以参考官网，有了rabbitmq_management 插件确实方便了不少。

下面就开始创建客户端进行测试了。

7.测试

创建两个.Java 类型的应用程序，一个用于发送消息，一个用于接收消息。

7.1创建一个maven项目，添加amqp-client依赖

<!--amqp-client-->
<dependency>
     <groupId>com.rabbitmq</groupId>
     <artifactId>amqp-client</artifactId>
     <version>5.7.3</version>
</dependency>
<!--slf4j日志-->
<dependency>
    <groupId>org.slf4j</groupId>
     <artifactId>slf4j-nop</artifactId>
     <version>1.7.2</version>
 </dependency>

7.2创建一个连接工具类，建立与RabbitMQ的连接

package demo.util;

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

/**
 * 连接工具类
 * @author qzz
 */
public class ConnectionUtil {

    /**
     * 建立与RabbitMQ的连接
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception{
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("localhost");
        //端口
        factory.setPort(5672);
        //设置账号信息  vhost(虚拟机)、 用户名、密码
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        //通过工厂获取连接
        Connection conn = factory.newConnection();
        return conn;
    }
}

7.3创建一个类，用于生产者发送消息

package demo;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import demo.util.ConnectionUtil;

/**
 * 生产者发送消息
 * @author qzz
 */
public class Send {

    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws Exception {

        //1.获取到连接
        Connection connection = ConnectionUtil.getConnection();
        //2.从连接中创建通道，使用通道才能完成与消息相关的操作
        Channel channel = connection.createChannel();
        //3.声明（创建）队列
        //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
        /**
         * 参数说明
         *   1.queue 队列名称
         *   2.durable 是否持久化；如果持久化，mq重启后队列还在
         *   3.exclusive 表示该消息队列是否只在当前connection生效（如果connection 连接关闭，则列队自动删除，如果将此参数设置为true可以用于临时队列的创建）
         *   4.autoDelete 自动删除，队列不再使用时是否自动删除此队列；如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
         *   5.arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
         */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //4.消息内容
        String message = "Hello World!";
        //5.向指定的队列中发送消息
        //参数：String exchange,String routingKey,BasicProperties props,bytes[]body
        /**
         * 参数说明：
         *   1.exchange 交换机，如果不指定将使用mq的默认交换机（设置为“”）
         *   2.routingKey 路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名字
         *   3.props 消息的属性
         *   4.body 消息内容
         */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("[X] send '" + message + "'");
        //6.关闭通道和连接 (资源关闭最好用try-catch-finally 语句处理)
        channel.close();
        connection.close();

    }
}

运行此类，控制台信息如下，生产者发送信息成功：
在这里插入图片描述
web管理页面：服务器地址/端口号（本地：http://localhost:15672，默认用户及密码：guest guest）

7.4创建一个类，用于消费者接收消息

package demo;

import com.rabbitmq.client.*;
import demo.util.ConnectionUtil;

import java.io.IOException;

/**
 * 消费者接受消息
 * @author qzz
 */
public class Receive {

    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws Exception{
        //1.获取到连接
        Connection connection = ConnectionUtil.getConnection();
        //2.从连接中创建会话通道，生成者金额mq服务所有通信都在channel通道中完成
        Channel channel = connection.createChannel();
        //3.声明（创建）队列
        //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
        /**
         * 参数说明
         *   1.queue 队列名称
         *   2.durable 是否持久化；如果持久化，mq重启后队列还在
         *   3.exclusive 表示该消息队列是否只在当前connection生效（如果connection 连接关闭，则列队自动删除，如果将此参数设置为true可以用于临时队列的创建）
         *   4.autoDelete 自动删除，队列不再使用时是否自动删除此队列；如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
         *   5.arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
         */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //4.实现消费方法
        DefaultConsumer consumer = 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();
                //body 即消息体
                String msg = new String(body,"utf-8");
                System.out.println("[X] received: " + msg);
            }
        };
        //5.监听队列，第二个参数：是否自动进行消息确认
        //参数：String queue ,boolean autoAck,Consumer callback
        /**
         * 参数说明：
         *   1.queue 队列名称
         *   2.autoAck 自动回复，当消费者接受到消息后要告诉mq消息已接收，如果将此参数设置为true表示会自动回复mq,如果设置为false要通过编程实现回复
         *   3.callback 消费方法，当消费者接收到消息时要执行的方法
         */
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}