RabbitMQ 消息队列学习

Posted 2022-05-17 IT_Holmes

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ 消息队列学习相关的知识，希望对你有一定的参考价值。

文章目录

1. MQ 消息队列
2. RabbitMQ 的四大核心概念
3. 各个名词的理解
4. RabbitMQ 安装
5. 安装rabbitmq的web页面插件
6. rabbitmq 添加用户并设置权限
7. 配置rabbitmq的java开发环境
8. 构建模拟 P(生产者) 代码
9. 构建模拟 C(消费者) 代码
10. 工作队列(Work Queues)

1. MQ 消息队列

1.1 MQ的概念

MQ(message queue) , 本质是个队列，FIFO先进先出的规则。队列中存放的是message内容，还是一种跨进程的通信机制。

MQ用于上下游传递消息：

上下游传递消息意思就是 A向B发送消息，那么A就是上游，B就是下游。
使用了MQ，上游只需要依赖MQ就可以了。

1.2 为什么要用MQ？

MQ的三大功能：

流量消峰：如果系统服务器1秒只能处理1万次请求，然后此时突然服务器接受到1万多次请求，该服务器就很有可能宕机了。

因此，就在中间加了个中间件MQ，目的就是为了做缓冲，减少服务器压力。

应用解耦：

以电商系统为例：

异步处理：

A想要B处理的数据，并且B还需要一段时间来处理。A就不用等待B处理，等B执行完后直接发送MQ，让MQ发给A就可以了。

1.3 MQ分类

ActiveMQ：

Kafka：

RocketMQ:

RabbitMQ:

1.4 MQ的选择

Kafka：比较适合大量数据的互联网服务的数据收集业务。大型公司建议选用，如果有日志采集功能，就要用到kafka了。

RocketMQ：天生为金融互联网领域而生，尤其是电商里面的订单扣款等等。RoketMQ在阿里双11经历了多次考验，因此对于像电商金融领域，可以考虑RocketMQ。

RabbitMQ：结合erlang语言本身的并发优势，性能好时效性微妙级。如果你的数据量没有那么大，中小型公司优先选择功能比较完备的RabbitMQ。

2. RabbitMQ 的四大核心概念

RabbitMQ只负责接受，存储和转发消息数据。并不会处理。

RabbitMQ的过程就像送快递一样：

RabbitMQ的四大核心概念：

生产者：发送消息。
交换机：绑定多个队列。
队列和消费者：一般一个队列对应一个消费者，也可以对应多个消费者，但是最终只有一个消费者能够拿到消息信息。

3. 各个名词的理解

Producer：生产者。

Connection：publisher/consumer和broker 之间的TCP连接，因为创建连接开销比较大，因此创建一次连接就可以，连接中有很多Channel。

Channel(频道)：

Broker：接受和分发消息的应用，RabbitMQ Server就是Message Broker。

Exchange：交换机，根据分发规则，将消息分发到队里中去。

Consumer：消费者。

4. RabbitMQ 安装

4.1 下载erlong安装包和rabbitmq安装包

因为，rabbitMQ是基于erlang语言的。因此，必须先要安装erlang环境。

你官方或者github上面下载rabbitMQ和erlang的包，之后在linux系统上解压：

erlong官方可以下载不同版本。

rabbitMq官方也可以去下载不同版本。

一般选择github下载，如下图：
下载地址:https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.8.8 。

rabbitmq和erlang有严格的版本对应，这一点一定注意，版本对应可以查看官方https://www.rabbitmq.com/which-erlang.html#intro

4.2 安装erlang 、安装rabbitmq 和安装socat

这里用的是ubuntu系统！

安装erlang：

#解压
tar -zxvf otp_src_21.3.1.tar.gz
#进入解压出来的otp_src_22.1目录
cd otp_src_21.3.1
# 配置并检查当前依赖库
# --prefix参数时安装路径，我的是/opt/erlang_22.1，如果是其它路径，需要事先创建好，比如我的这个路径创建命令：sudo mkdir /opt/erlang_22.1
./configure --prefix=/opt/rabbitMQ/erlang_21.3.1/ --with-ssl --enable-threads --enable-smp-support --enable-kernel-poll --enable-hipe --without-javac

执行上面命令出现下面就执行成功：

#编译
make
#安装
make install
#检验安装是否完成，上面设置了安装路径是/opt/erlang_22.1
/opt/rabbitMQ/erlang_21.3.1/bin/erl
#两次ctrl+c可以退出erl的编辑页面
#在/usr/local/bin下创建erl的软连接，注意安装目录，如果不创建软链接，rabbitmq启动时可能会报erl is not found
ln -s /opt/rabbitMQ/erlang_21.3.1/bin/erl /usr/local/bin/erl
#然后查看软链接是否生效，输入erl看能否进入erlang的编辑页面
erl


#如果已经有了erl，想要卸载步骤如下：
which erl
rm -rf /usr/bin/erl
rm -rf /usr/lib/erlang

rabbitmq依赖于erlong和socat，因此还要安装socat：

apt install -y socat

安装rabbitmq：

# 解压rabbitmq安装包
tar -xvf rabbitmq-server-generic-unix-3.8.8.tar.xz

# 启动rabbitmq服务
rabbitmq_server-3.8.8/sbin/rabbitmq-server

4.3 常用命令

配置rabbitmq的环境变量：

# 配置rabbitmq环境变量到profile中
vim /etc/profile

export PATH=$PATH:/opt/rabbitMQ/rabbitmq_server-3.8.8/sbin
export RABBITMQ_HOME=/opt/rabbitMQ/rabbitmq_server-3.8.8

# 配置好后，刷新一ixa
source /etc/profile

一个添加开机启动的命令：

# 添加开机启动rabbitmq服务 ，chkconfig命令可以用来设置、检查系统的各种服务。
# 注意：ubuntu上默认是不支持chkconfig命令。
chkconfig rabbitmq-server on

启动rabbitmq：

# 后台启动rabbitmq，detached是单独的，分离的意思。
rabbitmq-server -detached 
# 查看状态
rabbitmqctl status 
# 查看rabbitmq详情，注意grep后面的MQ大写或者只写rabbit
ps -aux | grep rabbitMQ
# 关闭可以直接kill -9 进程号

可能有些安装是直接将rabbitmq安装到了服务上面：

# 直接通过/sbin/service来运行，service是一个脚本文件可以运行服务中的一些内容。
/sbin/service rabbitmq-server start
/sbin/service rabbitmq-server status
/sbin/service rabbitmq-server stop

5. 安装rabbitmq的web页面插件

首先，停止rabbitmq服务。

# 关闭rabbitmq服务
kill -9 进程号
#或者
systemctl stop rabbitmq-server
#或者
/sbin/service rabbitmq-server stop

# 开启web管理插件
rabbitmq-plugins enable rabbitmq_management

这样我们再次开启rabbitmq服务：

并且访问打开端口号为15672的网站(远程网站要设置安全规则)。
开始的默认用户：guest，密码：guest。

所以要添加一个新的用户来登录web。

6. rabbitmq 添加用户并设置权限

# 创建账户
rabbitmqctl add_user admin 123

# 设置用户角色：rabbitmqctl set_user_tags [那个用户] [什么角色]
rabbitmqctl set_user_tags admin administrator

# 设置用户权限：
# rabbitmqctl set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

# 查看当前有哪些用户和角色
rabbitmqctl list_users

这样就创建好了一个超级管理员用户。

这样就可以用我们设置的用户来登录了。

在web页面，admin中能看到用户相关数据：

完成上面步骤rabbitmq服务就安装完成了。

因此，上面红色部分就是RabbitMQ代表使用者保留的消息缓冲区。

接下来开始完成P(生产者)，C(消费者)。

7. 配置rabbitmq的java开发环境

第一步：创建一个项目。

第二步：配置jdk编译版本，两种方式：

<!--方式一：指定jdk编译版本。如果不指明版本就容易出现版本不匹配的问题，可能导致编译不通过的问题。-->
<properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
</properties>

<!--方式二：指定jdk编译版本-->
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>8</source>
                <target>8</target>
            </configuration>
        </plugin>
    </plugins>
</build>

第三步：导入rabbimq客户端依赖和 commons-io依赖。

<dependencies>
    <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.8.0</version>
    </dependency>
    
    <dependency>
        <groupId>commons-io</groupId>
        <artifactId>commons-io</artifactId>
        <version>2.6</version>
    </dependency>
</dependencies>

8. 构建模拟 P(生产者) 代码

生产者是发送消息给中间件队列(这里是rabbitmq)，之后队列再给消费者。

构建消息生产者：

注意：发消息之前也要开启5672端口，设置好白名单！
5672端口是rabbitmq 默认TCP监听端口

package com.itholmes.rabbitmq.one;

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

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

/**
 * 生产者代码
 */
public class Producer 

    //队列名称
    public static final String QUEUE_NAME = "hello";

    //发消息
    public static void main(String[] args) throws IOException, TimeoutException 

        /**
         *  创建一个rabbitMq的连接工厂
         */
        ConnectionFactory factory = new ConnectionFactory();

        //设置工厂IP(连接RabbitMQ队列的对应IP)
        factory.setHost("39.103.163.156");
        //rabbitmq对应的用户名
        factory.setUsername("admin");
        //密码
        factory.setPassword("0818");

        /**
         *  创建一个rabbitMq的连接
         */
        //创建连接
        Connection connection = factory.newConnection();
        //之前rabbitmq原理里面，一个连接中会有多个channel信道。因此要创建信道。
        //获取信道
        Channel channel = connection.createChannel();

        /**
         *  生成一个队列
         *      channel.queueDeclare()的参数：(声明队列方法的参数)
         *          参数1：队列名称
         *          参数2：队列里面的消息是否持久化(持久化存储在磁盘中)，默认情况消息存储在内存中。
         *          参数3：该队列是否只供一个消费者进行消费，是否进行消息共享。true可以多个消费者消费。默认是不允许多个消费者消费。
         *          参数4：是否自动删除，最后一个消费者断开连接以后，该队列是否自动删除。true自动删除，false不自动删除。
         *          参数5：其他参数。
         */
        //声明一个队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //发消息
        String message = "hello world";

        /**
         *   basicPublish的publish就有发布的意思。
         *      channel.basicPublish()的参数：
         *          参数1：发送到哪个交换机。
         *          参数2：路由的key值是哪个，本次是队列的名称。
         *          参数3：其他参数信息。
         *          参数4：发送消息的消息体。
         */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("消息发送完毕");

9. 构建模拟 C(消费者) 代码

消费者代码：

package com.itholmes.rabbitmq.one;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 *  消费者代码
 */
public class Consumer 

    //队列的名称
    public static final String QUEUE_NAME = "hello";

    //接受消息
    public static void main(String[] args) throws IOException, TimeoutException 

        //创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("39.103.163.156");
        factory.setUsername("admin");
        factory.setPassword("0818");

        //创建一个连接
        Connection connection = factory.newConnection();
        //创建一个信道
        Channel channel = connection.createChannel();

        /**
         *  声明下面参数3的 回调函数
         *      参数3：声明接受消息的回调
         *          (String var1, Delivery var2)-> 可以缩写为 (var1,var2)->：因为前面有接口对应上了。
         */
        DeliverCallback deliverCallback = (String var1, Delivery var2) -> 
            //这里的var2就是message消息。 getBody()是获取var2的消息体。
            System.out.println(new String(var2.getBody()));
        ;

        /**
         *  声明下面参数4的 回调函数
         *      消费者取消消费(消费者中断消费)要调用的回调
         */
        CancelCallback cancelCallback = (String var1)->
            System.out.println("消费者消费被中断");
        ;

        /**
         *  消费者接受消息(也可以理解为消费者消费消息)
         *      channel.basicConsume()方法参数对应：
         *          参数1：消费那个队列（队列名）
         *          参数2：消费成功之后是否要自动应答，true代表的自动应答，false代表手动应答。
         *          参数3：消费者声明接受消息的回调(疑惑：未成功消费要调用的回调)。
         *          参数4：消费者取消消费(消费者中断消费)要调用的回调。
         */
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);

这样就会接受rabbitmq服务端的消息了。前提是队列中有消息。

10. 工作队列(Work Queues)

10.1 工作队列原理介绍

工作队列又称为任务队列。

意思就是一个生产者发送大量消息到队列，由多个工作线程进行接收。

主要思想是为了解决密集型任务：

当有多个工作线程，这些工作线程将一起处理这些任务。
分发策略是：轮询分发消息(依次分发)。

10.2 封装Utils工具类

因为有很多代码是重复的，所以封装了一个工具类：

package com.itholmes.rabbitmq.utils;

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

/**
 *  操作rabbitmq的工具类
 *      获取一哥channel信道
 */
public class RabbitMqUtils 

    private static String IP = "39.103.163.156";
    private static String USERNAME = "admin";
    private static String PASSWORD = "0818";

    //得到一个连接的channel
    public static Channel getChannel() throws IOException, TimeoutException 
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(IP);
        factory.setUsername(USERNAME);
        factory.setPassword(PASSWORD);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;

10.3 模拟生产者 + 多个工作线程的过程

写两个或者多个工作线程：

可以写两个或者多个main方法，也可以写一个main方法执行两次或者多次(注意下图)！

工作线程代码：

package com.itholmes.rabbitmq.two;

import com.itholmes.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 *  这是一个工作线程(相当于消费者)
 */
public class Worker01 

    //队列的名称
    public static final String QUEUE_NAME = "hello";

    //接受消息
    public static void main(String[] args) throws IOException, TimeoutException 
        Channel channel = RabbitMqUtils.getChannel();

        DeliverCallback deliverCallback = (var1, var2)->
            System.out.println("work01接受到的消息："+var2.getBody());
        ;

        CancelCallback cancelCallback = (var1)->
            System.out.println(var1+"消息者取消消费接口回调逻辑");
        ;

        System.out.println("C1等待接受消息...");
        //消息的接受
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);

生产者代码：

模拟大量发送消息过程。

package com.itholmes.rabbitmq.two;

import com.itholmes.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;

/**
 *  生产者发送大量消息
 */
public class Producer 
    //队列名称
    public static final String QUEUE_NAME = "hello";
    //发送大量消息
    public static void main(String[] args) throws IOException, TimeoutException 
        Channel channel = RabbitMqUtils.getChannel();
        //队列的声明
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //发送消息，模拟从控制台中输入接受消息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext())
            String message = scanner.next();
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("发送消息完成："+message);

这样启动测试，就会发现他是轮询将消息分发给多个工作线程的。

以上是关于RabbitMQ 消息队列学习的主要内容，如果未能解决你的问题，请参考以下文章

RABBITMQ初探——消息分发

python 学习总结3 消息队列RabbitMQ

Java开发 - 消息队列之RabbitMQ初体验

了解MQ和安装使用RabbitMQ