RabbitMQ 消息队列学习 (一)

文章目录

• 1. MQ 消息队列
• • 1.1 MQ的概念
  • 1.2 为什么要用MQ？
  • 1.3 MQ分类
  • 1.4 MQ的选择
• 2. RabbitMQ 的四大核心概念
• 3. 各个名词的理解
• 4. RabbitMQ 安装
• • 4.1 下载erlong安装包和rabbitmq安装包
  • 4.2 安装erlang 、 安装rabbitmq 和 安装socat
  • 4.3 常用命令
• 5. 安装rabbitmq的web页面插件
• 6. rabbitmq 添加用户并设置权限
• 7. 配置rabbitmq的java开发环境
• 8. 构建模拟 P(生产者) 代码
• 9. 构建模拟 C(消费者) 代码
• 10. 工作队列(Work Queues)
• • 10.1 工作队列原理 介绍
  • 10.2 封装Utils工具类
  • 10.3 模拟生产者 + 多个工作线程的过程

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/profileexport 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 ]    
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"# 查看当前有哪些用户和角色
rabbitmqctl list_users

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

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

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

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

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

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

7. 配置rabbitmq的java开发环境

第一步：创建一个项目。

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

<properties><maven.compiler.source>8maven.compiler.source><maven.compiler.target>8maven.compiler.target>
properties>
<build><plugins><plugin><groupId>org.apache.maven.pluginsgroupId><artifactId>maven-compiler-pluginartifactId><configuration><source>8source><target>8target>configuration>plugin>plugins>
build>

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

<dependencies><dependency><groupId>com.rabbitmqgroupId><artifactId>amqp-clientartifactId><version>5.8.0version>dependency><dependency><groupId>commons-iogroupId><artifactId>commons-ioartifactId><version>2.6version>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";//得到一个连接的channelpublic 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);}}}

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

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！