RabbitMQ(一) 安装配置与简单使用

2023-03-10 阅读量

参考文章

一使用docker安装

RabbitMQ+docker安装教程

二手动安装

RabbitMQ是一个开源的遵循 AMQP协议实现的基于 Erlang语言编写，即需要先安装部署Erlang环境再安装RabbitMQ环境

RabbitMQ和Erlang的版本对应关系参考:RabbitMQ Erlang Version Requirements — RabbitMQ

下面以CentOS7安装RabbitMQ-3.8.13为例

1 安装Erlang

下载Erlang-22.3.4.12

1	wget --content-disposition https://packagecloud.io/rabbitmq/erlang/packages/el/7/erlang-22.3.4.12-1.el7.x86_64.rpm/download.rpm

安装

1	yum localinstall erlang-22.3.4.12-1.el7.x86_64.rpm

安装完成

2 安装rabbitMQ

下载rabbitMQ-3.8.13

1	wget --content-disposition https://packagecloud.io/rabbitmq/rabbitmq-server/packages/el/7/rabbitmq-server-3.8.13-1.el7.noarch.rpm/download.rpm

安装

1 2	rpm --import https://www.rabbitmq.com/rabbitmq-release-signing-key.asc yum localinstall rabbitmq-server-3.8.13-1.el7.noarch.rpm

安装完成

三启动关闭和重启

启动rabbitMQ默认端口**5672**

1	systemctl start rabbitmq-server

关闭rabbitMQ

1	systemctl stop rabbitmq-server

重启rabbitMQ

1	systemctl restart rabbitmq-server

查看rabbitMQ启动状态

1	systemctl status rabbitmq-server

设置开机自启动命令

1	systemctl enable rabbitmq-server

四 RabbitMQ Web界面管理

安装web客户端管理插件

1	rabbitmq-plugins enable rabbitmq_management

重启服务即可

配置防火墙开放15672端口

rabbitmq有一个默认账号和密码是:guest 。默认情况只能在 localhost本机下访问（见下图），所以需要新增一个远程登录的用户

1	rabbitmqctl add_user 账号密码

添加成功

设置用户分配操作权限

1	rabbitmqctl set_user_tags 账号 administrator

通过http://IP地址:15672 进行Web页面登录，输入账号密码，完成页面访问

五 MQ集群搭建

参考RabbitMQ集群搭建 - 腾讯云开发者社区-腾讯云 (tencent.com)

六认识RabbitMQ Management

在RabbitMQ Management界面可以看到 RabbitMQ 中的一些基本概念，例如每个菜单分别是：

Override：整体一览。
Connections：网络连接，比如一个 TCP 连接。
Channels：信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的 TCP 连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
Exchanges：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Queues：消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

具体后续文章详细学习

在这里进行一些配置方便后续测试

1 创建虚拟主机

2 检查权限

检查刚刚创建的用户是否可以访问该虚拟主机

没有权限点击红框处添加对应虚拟主机

七简单示例

引入maven依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.7.3</version>
</dependency>

1 RabbitMQProducer

创建 RabbitMQProducer 类，使用 Channel 发送消息。代码如下：

public class RabbitMQProducer {

    private static final String IP_ADDRESS = "ip地址";
    private static final Integer PORT = 5672; //默认端口
    private static final String USERNAME = "账号";
    private static final String PASSWORD = "密码";

    private static final String EXCHANGE_NAME = "exchange_demo";
    private static final String ROUTING_KEY = "routingkey_demo";
    /**
     * 只有 QUEUE_NAME 需要共享给 RabbitMQConsumer
     */
    public static final String QUEUE_NAME = "queue_demo";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        Connection connection = getConnection();

        // 创建信道
        Channel channel = connection.createChannel();

        // 初始化测试用的 Exchange 和 Queue
        initExchangeAndQueue(channel);

        // 发送 3 条消息
        for (int i = 0; i < 3; i++) {
            String message = "Hello World" + i;
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
        }

        // 关闭
        channel.close();
        connection.close();
    }

    public static Connection getConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(IP_ADDRESS);
        factory.setPort(PORT);
        factory.setUsername(USERNAME);
        factory.setPassword(PASSWORD);
        factory.setVirtualHost("/test");
        return factory.newConnection();
    }

    // 创建 RabbitMQ Exchange 和 Queue ，然后使用 ROUTING_KEY 路由键将两者绑定。
    // 该步骤，其实可以在 RabbitMQ Management 上操作，并不一定需要在代码中
    private static void initExchangeAndQueue(Channel channel) throws IOException {
        // 创建交换器：direct、持久化、不自动删除
        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);

        // 创建队列：持久化、非排他、非自动删除的队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 将交换器与队列通过路由键绑定
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
    }

}

运行测试后发现队列中已有三条测试发送的消息

2 RabbitMQConsumer

创建 RabbitMQConsumer类，使用 Channel 发送消息。代码如下：

// RabbitMQConsumer.java

public class RabbitMQConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        Connection connection = RabbitMQProducer.getConnection();

        // 创建信道
        final Channel channel = connection.createChannel();
        channel.basicQos(64); // 设置客户端最多接收未被 ack 的消息数量为 64 。

        // 创建消费者
        Consumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 打印日志
                System.out.println(String.format("[线程：%s][路由键：%s][消息内容：%s]",
                        Thread.currentThread(), envelope.getRoutingKey(), new String(body)));
                // ack 消息已经消费
                channel.basicAck(envelope.getDeliveryTag(), false);
            }

        };
        // 订阅消费 QUEUE_NAME 队列
        channel.basicConsume(RabbitMQProducer.QUEUE_NAME, consumer);

        // 关闭
        try {
            TimeUnit.HOURS.sleep(1);
        } catch (InterruptedException ignore) {
        }
        channel.close();
        connection.close();
    }

}

可以看到刚刚的消息被成功消费了

一 使用docker安装

二 手动安装