RabbitMQ学习笔记(一)

最近马上面临过年，就等着下周发年终奖了哈哈，先给大家拜个早年，然后再一起来学习RabbitMQ。

RabbitMQ是一个消息队列中间件。

中间件是处于操作系统和应用程序之间的软件，一般用于跨平台，跨应用，跨语言的信息互通，持久化等操作，只要通信协议相同即可。

消息队列中间件是一个应用于分布式架构中的传递信息数据进行数据收集与分发的一个中间件。一般用于如下场景：

1.跨系统数据传递

2.高并发的流量削峰

3.数据分发与异步处理

4.大数据分析与传递

5.分布式事务

常见的消息中间件有：

名称	特点
ActiveMQ	比较老牌的消息中间件，比较复杂
RabbitMQ	与Spring同厂商，在普遍SpringBoot框架下RabbitMQ的发挥空间肯定是最大的
Kafka	高性能中间件，但不支持事务
RocketMQ

消息队列通信协议

消息队列需要完成跨应用，跨平台的通信，信息的传递，那么他就要指定一个大家都要遵守的协议，才能完成信息的传递，底层的协议分为TCP与UDP协议，如果有不懂的同学可以自己去查询一下，稍微理解依稀概念就好，但是这种底层的协议不能满足我们消息队列功能的构建，他可能只约定了一部分功能，所以我们需要在底层协议上自己添加限制内容，使消息队列能够完成更加复杂的工作。

常见的消息中间件协议有：OpenWire、AMQP、MQTT、Kafka、OpenMessage协议等。

协议名称	特点
AMQP协议	RabbitMQ(默认)、ActiveMQ支持 1.分布式事务支持 2.消息持久化支持 3.高性能和高可靠的消息处理
MQTT协议	RabbitMQ(需要开启)、ActiveMQ支持是物联网系统架构的重要组成部分 1.轻量级 2.结构简单 3.传输快，不支持事务 3.没有持久化设计
OpenMessage协议	RocketMQ支持 1.结构简单 2.解析速度快 3.支持事务及持久化设计
Kafka协议	Kafka支持基于TCP/IP协议的二进制协议，速度最快 1.结构简单 2.解析速度快 3.无事务支持 3.有持久化设计

RabbitMQ采用了AMQP协议，是用Erlang语言开发的消息中间件。

面试1：为什么消息队列发送消息不直接采用http协议呢？

答：因为http协议的请求头与请求体内容复杂，包括cookie，数据的加解密，状态码，响应码等功能，但是对于消息队列来说，我们不要求这些，我们只要求能够简洁快速的传递信息，存储数据，分发消息就可以了。

其次http大部分为短连接，在交互过程中难免面临响应中断，不会进行持久化，造成消息丢失。这是消息中间件不能忽略的问题，所以http协议并不适用于消息中间件。

消息分发策略机制对比

	ActiveMQ	RabbitMQ	Kafka	RocketMQ
发布订阅	支持	支持	支持	支持
轮询分发	支持	支持	支持	不支持
公平分发	不支持	支持	支持	不支持
重发	支持	支持	不支持	支持
消息拉取	不支持	支持	支持	支持

RabbitMQ下载安装

RabbitMQ角色分类

RabbitMQ安装过后，我们可以通过IP地址加端口号进行访问。同时可以注册用户，注册的用户需要分配权限，则RabbitMQ有如下几个角色权限。

角色名称	权限
none	无法登录的权限
management	只能查看与自己有关的节点
policymaker	可以查看、删除、创建自己的节点
monitoring	可以查看、删除、创建自己的节点外，还可以查看其他人的节点
administrator	最高权限，可以查看、删除、创建所有节点，并且可以创建用户，赋予权限

RabbitMQ简单实战篇 Maven项目

先导入依赖


    com.rabbitmq
    amqp-client
    5.10.0

创建生产者

package com.test.rabbitmq.simple;

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 void main(String[] args) throws IOException, TimeoutException {
        // 创建连接对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 放入要访问的RabbitMQ的IP地址
        connectionFactory.setHost("127.0.0.1");
        // 放入访问端口号
        connectionFactory.setPort(5672);
        // 放入登录RabbitMQ的用户账号
        connectionFactory.setUsername("admin");
        // 放入用户密码
        connectionFactory.setPassword("admin");
        // 配置虚拟访问节点，把消息发往根节点下
        // 节点的作用是用于隔离，例如C盘，D盘概念，如果有多种多样的消息放在一起会杂乱
        connectionFactory.setVirtualHost("/");
        // 创建连接
        Connection connection = connectionFactory.newConnection("连接名称");
        // 通过连接获取通道channel
        Channel channel = connection.createChannel();
        // 进行队列声明，参数含义：1.队列名称，2.是否要持久化，3.是否要有排他性，4.发送以后是否要自动删除，5。是否携带其他参数
        // 持久化表示，随着RabbitMQ服务器重启，原来创建的队列是否还会存在，非持久化会存盘，但是会随着服务器重启而丢失
        // 是否为独占队列
        // 随着队列中最后一个消息被消费，是否自动删除队列
        channel.queueDeclare("队列名称", false, false, false, null);
        // 配置发送的消息
        String str = "Hello world";
        // 发送消息给队列，参数含义：1.交换机(虽然为空也有默认交换机，但是工作中最好指定交换机)，2.队列名称/路由key，3.消息状态控制，输入参数后用于headers模式进行发送消息，4.消息内容(为字节)
        channel.basicPublish("", "队列名称", null, str.getBytes());
        // 关闭通道
        if (channel != null && channel.isOpen()) {
            channel.close();
        }
        // 关闭连接
        if (connection != null && connection.isOpen()) {
            connection.close();
        }
    }
}

创建消费者

package com.test.rabbitmq.simple;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

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


public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 放入要访问的RabbitMQ的IP地址
        connectionFactory.setHost("127.0.0.1");
        // 放入访问端口号
        connectionFactory.setPort(5672);
        // 放入登录RabbitMQ的用户账号
        connectionFactory.setUsername("admin");
        // 放入用户密码
        connectionFactory.setPassword("admin");
        // 配置虚拟访问节点，把消息发往根节点下
        connectionFactory.setVirtualHost("/");
        // 创建连接
        Connection connection = connectionFactory.newConnection("连接名称");
        // 通过连接获取通道channel
        Channel channel = connection.createChannel();
        // 接收队列消息，其中队列名称为消费哪条队列的消息则使用哪条队列的名称
        // 第二个参数为是否自动应答，一般为false，因为true会导致死循环
        // false关闭自动应答后进行手动应答，如果不懂请继续向下看，work模式中有讲解
        // 手动应答使用 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        channel.basicConsume("队列名称", true, new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                // 消息成功接受时
                System.out.println("收到的消息" + new String(delivery.getBody(), StandardCharsets.UTF_8));
            }
        }, new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {
                // 出现异常的时候应该怎么处理
                System.out.println("接收失败了");
            }
        });
        
        // 关闭通道
        if (channel != null && channel.isOpen()) {
            channel.close();
        }
        // 关闭连接
        if (connection != null && connection.isOpen()) {
            connection.close();
        }
    }
}

面试2：为什么消息队列的生产与消费是基于通道，而不是连接的呢？

答：因为rabbitMQ是基于TCP/IP协议的，需要经历三次握手跟四次挥手。所以连接的每一次创建与关闭都会十分消耗性能，但是通道则不会。每一个连接中都会有多个通道，基于通道的消息性能会特别高。

核心概念：

名称	含义
Server/Broker	接收客户端的连接，实现AMQP实体服务，安装rabbitmq-server
Connection(连接)	应用程序与Broker的网络连接
Channel(网络信息通道)	几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道，每一个Channel都代表一个会话任务
Message(消息)	服务与应用程序之间传送的数据，由Properties和body组成，Properties可对消息进行修饰，比如优先级、延时处理等；body则是消息体的内容
Virtual Host(虚拟地址)	用于进行逻辑隔离，同一个虚拟主机内可以有多个Exchange，但不能有名称相同的Exchange
Exchange(交换机)	接收消息，根据路由key发送消息到绑定的队列，不具备消息存储的能力
Bindings(虚拟连接)	Exchange与Queue之间的虚拟连接
Routing key(路由规则)	虚拟机通过Routing key来决定消息的发送与接收
Queue(队列)	消息队列，接收Exchange的消息保存到队列中，并转发给消费者，

RabbitMQ的消息发布模式

RabbitMQ消息发布模式

上面的网址可以查看RabbitMQ的全部消息发布模式。

1.简单模式

创建生产者与消费者，通过队列，生产者进行生产，消费者进行消费，如上图代码示例

2.Fanout模式

发布与订阅模式，下图中深蓝色X 表示交换机，我们需要创建一个交换机，然后绑定交换机与队列，这样我们向此交换机投递消息的时候，绑定到的队列全部都会收到消息。设置路由Key没有意义。

3.Direct模式

路由模式，在Fanout模式之上，添加了路由Key，如果设置路由Key，就会指定发送到对应路由key的队列中，进一步缩小消息发送的范围。

上面简单模式创建的生产者使用的为默认的交换机，那么我们应该如何使用代码进行交换机的创建及与队列绑定呢？代码如下：

package com.test.rabbitmq.direct;

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 void main(String[] args) throws IOException, TimeoutException {
        // 创建连接对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 放入要访问的RabbitMQ的IP地址
        connectionFactory.setHost("127.0.0.1");
        // 放入访问端口号
        connectionFactory.setPort(5672);
        // 放入登录RabbitMQ的用户账号
        connectionFactory.setUsername("admin");
        // 放入用户密码
        connectionFactory.setPassword("admin");
        // 配置虚拟访问节点，把消息发往根节点下
        connectionFactory.setVirtualHost("/");
        // 创建连接
        Connection connection = connectionFactory.newConnection("连接名称");
        // 通过连接获取通道channel
        Channel channel = connection.createChannel();
        // 交换机声明，参数含义：1.交换机名称，2.交换机类型，3.是否持久化，表示服务器重启后交换机是否被移除
        channel.exchangeDeclare("交换机名称", "交换机类型", true);
        // 进行队列声明，参数含义：1.队列名称，2.是否要持久化，3.是否要有排他性，4.发送以后是否要自动删除，5。是否携带其他参数，Header模式使用
        // 持久化表示，随着RabbitMQ服务器重启，原来创建的队列是否还会存在，非持久化会存盘，但是会随着服务器重启而丢失
        // 是否为独占队列
        // 随着队列中最后一个消息被消费，是否自动删除队列
        channel.queueDeclare("队列名称1", false, false, false, null);
        channel.queueDeclare("队列名称2", false, false, false, null);
        channel.queueDeclare("队列名称3", false, false, false, null);
        // 进行绑定队列到指定交换机上
        channel.queueBind("队列名称1", "交换机名称", "Routing Key1");
        channel.queueBind("队列名称2", "交换机名称", "Routing Key2");
        channel.queueBind("队列名称3", "交换机名称", "Routing Key3");
        // 配置发送的消息
        String str = "Hello world";
        // 发送消息给队列
        channel.basicPublish("指定使用的交换机名称", "路由名称", null, str.getBytes());
        // 关闭通道
        if (channel != null && channel.isOpen()) {
            channel.close();
        }
        // 关闭连接
        if (connection != null && connection.isOpen()) {
            connection.close();
        }
    }
}

4.Topic模式

主题模式，在路由模式的基础上可以模糊匹配路由，路由可以为xxxx.xxxx，由如下符号进行模糊匹配变为 *.xxxx或#.xxxx.* 等多种类型含义。

符号	含义
*	必须匹配一级
#	匹配0级或多级

5.Headers模式

参数模式，在页面中没有模型，概念为我们创建交换机的时候添加参数，之后我们创建channel.basicPublish对象时第三个参数附带值后，根据值内容来对相应的队列进行消息投递。与路由模式类似。

6.Work模式

工作模式，其中由两种模式，分别为：轮询模式、公平分发模式。

轮询模式：每个消费者一条，不会多发也不会少发。

公平分发模式：能者多劳，处理的快就多消费，处理的慢就少消费。

确定轮询分发还是公平分发，两者的消费者代码是不同的，我分别示例

轮询分发代码：

package com.test.rabbitmq.work;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

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


public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 放入要访问的RabbitMQ的IP地址
        connectionFactory.setHost("127.0.0.1");
        // 放入访问端口号
        connectionFactory.setPort(5672);
        // 放入登录RabbitMQ的用户账号
        connectionFactory.setUsername("admin");
        // 放入用户密码
        connectionFactory.setPassword("admin");
        // 配置虚拟访问节点，把消息发往根节点下
        connectionFactory.setVirtualHost("/");
        // 创建连接
        Connection connection = connectionFactory.newConnection("连接名称");
        // 通过连接获取通道channel
        Channel channel = connection.createChannel();
        // 每次从队列中取多少条消息进行消费
//        channel.basicQos(1);
        // 接收队列消息，其中队列名称为消费哪条队列的消息则使用哪条队列的名称
        // 轮询分发的关键在于自动应答，第二个参数为true
        channel.basicConsume("队列名称", true, new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                // 消息成功接受时
                System.out.println("收到的消息" + new String(delivery.getBody(), StandardCharsets.UTF_8));
                // 手动应答
//                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }, new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {
                // 出现异常的时候应该怎么处理
                System.out.println("接收失败了");
            }
        });
        
        // 关闭通道
        if (channel != null && channel.isOpen()) {
            channel.close();
        }
        // 关闭连接
        if (connection != null && connection.isOpen()) {
            connection.close();
        }
    }
}

公平分发代码：

package com.test.rabbitmq.work;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

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


public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 放入要访问的RabbitMQ的IP地址
        connectionFactory.setHost("127.0.0.1");
        // 放入访问端口号
        connectionFactory.setPort(5672);
        // 放入登录RabbitMQ的用户账号
        connectionFactory.setUsername("admin");
        // 放入用户密码
        connectionFactory.setPassword("admin");
        // 配置虚拟访问节点，把消息发往根节点下
        connectionFactory.setVirtualHost("/");
        // 创建连接
        Connection connection = connectionFactory.newConnection("连接名称");
        // 通过连接获取通道channel
        Channel channel = connection.createChannel();
        // 每次从队列中取多少条消息进行消费
        // 公平分发时需要配置
        channel.basicQos(1);
        // 接收队列消息，其中队列名称为消费哪条队列的消息则使用哪条队列的名称
        // 公平分发的不可以自动应答，第二个参数为false
        channel.basicConsume("队列名称", false, new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                // 消息成功接受时
                System.out.println("收到的消息" + new String(delivery.getBody(), StandardCharsets.UTF_8));
                // 公平分发需要配置手动应答
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }, new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {
                // 出现异常的时候应该怎么处理
                System.out.println("接收失败了");
            }
        });
        
        // 关闭通道
        if (channel != null && channel.isOpen()) {
            channel.close();
        }
        // 关闭连接
        if (connection != null && connection.isOpen()) {
            connection.close();
        }
    }
}

下一篇请戳↓

RabbitMQ学习笔记(二)

RabbitMQ学习笔记(一)

大数据系统相关栏目本月热门文章