📒博客首页：崇尚学技术的科班人
🏇小肖来了
🍣今天给大家带来的文章是《消息队列的入门篇》🍣
🍣我们的项目中可能会用到这个吗？我的回答是：肯定的。🍣
🍣希望各位小伙伴们能够耐心的读完这篇文章🍣
🙏博主也在学习阶段，如若发现问题，请告知，非常感谢🙏
💗同时也非常感谢各位小伙伴们的支持💗

1、消息队列

1.1、MQ的相关概念

1、什么是MQ？

MQ，从字面意思来看就是个队列，遵循FIFO先入先出原则，只不过队列中存放的是消息，它还是一种跨进程的通信机制，用于上下游传递消息。
在互联网的架构中，MQ是一种非常常见的上下游**“逻辑解耦 + 物理解耦”**的消息通信服务。使用了MQ之后，消息发送上游只需要依赖MQ，不用依赖其他服务。

2、为什么要使用MQ？

流量消峰

图中的订单系统的最高限制是1万次/s。而如果我们添加了消息队列的话，它会对请求进行排队。但这在一定程度上会影响请求的速度。但总比不能下单好。

应用解耦

上图的左边部分中可以看出订单系统和其他各个系统直接相连，那么如果其他各个系统中的一个出现问题的话，整个系统都会宕机。而右边部分的话，订单系统通过MQ和其他系统进行联系。当订单系统处理完自己的事情之后，由MQ对其他系统进行分配任务和监督。所以如果其中的一个系统出现问题不会影响整个系统。

异步处理

由上图可知，有些服务之间的调用是异步的，例如A调用B。A需要知道B什么时候将服务执行完毕。此时A就需要调用B的回调API。但是这样做，对于A来说需要每时每刻进行调用。还有一种方法就是使用MQ。当B服务结束之后，B会向MQ发送消息。然后MQ会向A发送处理消息。A服务还能即使的得到处理成功的消息。

3、MQ的分类

ActiveMQ
kafka
RocketMQ
RabbitMQ

4、MQ的选择

1. kafka

kafka适合大量数据的互联网服务的数据收集业务。大型公司建议可以选用，如果有日志采集功能，肯定是首选kafka了。

2. RocketMQ

天生为金融互联网领域而生。如果你的业务有像双11的那样的高并发的场景，建议可以选择RocketMQ。

3. RabbitMQ

RabbitMQ结合erlang语言本身的并发优势，性能好时效性微秒级，社区活跃度也比较高，如果你的数据量没有那么大的话可以使用它。

1.2、RabbitMQ

1、RabbitMQ的概念

RabbitMQ是一个消息中间件；它接受并转发消息。它可以接收、存储和转发消息数据。

2、四大核心概念

生产者
交换机
队列
消费者

3、RabbitMQ核心部分

4、各个名词的介绍

Broker：接受和分发消息的应用，也就是消息队列。
Virtual Host：每一个消息队列都可以划分出多个vhost，每个用户在自己的vhost创建exchange/queue等。
connection：生产者或消费者和消息队列之间的TCP连接。
channel：消息信道，channel作为轻量级的connection极大减少了操作系统建立TCP连接的开销。
exchange：交换机，消息到达消息队列的第一个地方，它会根据分发规则，匹配查询表中的路由关键字，分发消息到队列中。
queue：消息最终被送到这里等待被消费者取走。
binding：交换机和队列之间的虚拟连接，绑定中可以包含如有关键字，绑定信息被保存到交换机的查询表中，用于消息的分发依据。

5、安装

从官网进行下载相应的安装文件
官网下载地址
将相关的安装文件上传到相应的/opt目录下

安装文件（按照以下顺序执行）

rpm -ivh erlang-22.3-1.el7.x86_64.rpm

yum install socat -y

rpm -ivh rabbitmq-server-3.8.8-1.el7.noarch.rpm

常用命令（按照以下顺序执行）

# 添加开机启动RabbitMQ服务
chkconfig rabbitmq-server on

# 启动服务
/sbin/service rabbitmq-server start

# 查看服务状态
/sbin/service rabbitmq-server status

# 停止服务（选择执行）
/sbin/service rabbitmq-server stop

安装并启动成功后的图片说明

开启web管理插件

rabbitmq-plugins enable rabbitmq_management

需要创建用户权限。

添加一个新的用户

# 创建账号
rabbitmqctl add_user admin 123

# 设置用户角色
rabbitmqctl set_user_tags admin administrator

# 设置用户权限
set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
#用户user_admin具有/vhost1这个virtual host 中的所有资源的配置、写、读权限。

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

登录成功图片说明

2、HelloWorld

“P”是我们的生产者，“C”是我们的消费者。中间的框是一个队列 - RabbitMQ代表使用者保留的消息缓冲区。

总的说明

2.1、引入的依赖

<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>

    <dependencies>
        <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>

2.2、生产者（发送消息）

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

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置IP地址
        factory.setHost("192.168.123.129");
        // 用户名
        factory.setUsername("admin");
        // 密码
        factory.setPassword("123");

        // 创建连接
        Connection connection = factory.newConnection();
        // 获取信道
        Channel channel = connection.createChannel();
        /** * 声明队列 * 1.队列名称 * 2.队列里面的消息是否持久化（磁盘）默认情况下消息存储在内存中 * 3.该队列是否提供一个消费者进行消费，就是是否进行消息共享 * 4.就是当最后一个消费者断开连接之后，该队列是否自动删除消息 * 5.其他参数 */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // 发消息
        String message = "hello,world!!";

        /** * 发送一个消息 * 1.发送到哪个交换机 * 2.路由的Key值，也就是本次队列的名称 * 3.其他参数信息 * 4.发送消息的消息内容 */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
        System.out.println("消息发送完毕");
    }
}

插件的观察

2.3、消费者（接收消息）

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("192.168.123.129");
        factory.setUsername("admin");
        factory.setPassword("123");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        DeliverCallback deliverCallback = (var1, var2)->{
            System.out.println(new String(var2.getBody()));
        };

        CancelCallback cancelCallback = var1->{
            System.out.println("消费消息被中断");
        };
        /** * 接收消息 * 1.消费哪个队列 * 2.消费成功之后是否要自动应答 * 3.消费者未成功消费的回调 * 4.消费者取消消费的回调 */
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

一文带你入门消息队列RabbitMQ