RabbitMQ系列RabbitMQ入门
Posted 一宿君
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ系列RabbitMQ入门相关的知识，希望对你有一定的参考价值。
RabbitMQ入门
学习目标
- 能够说出什么是消息队列
- 能够安装RabbitMQ
- 能够编写RabbitMQ的入门程序
- 能够说出RabbitMQ的5种模式特征
- 能够使用SpringBoot整合RabbitMQ
1. 消息队列概述
1.1. 消息队列MQ
MQ全称是Message Queue，消息队列是应用程序和应用程序之间的通信方法。
-
为什么使用MQ？
- 在项目中，可将一些无需即时返回且耗时的操作提取出来，进行异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。
-
开发中消息队列通常有如下应用场景：
- 任务异步处理：将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间。
- 应用程序解耦合：MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合。
1.2、AMQP 和 JMS
MQ是消息通信的模型；实现MQ的大致有两种主流方式：AMQP、JMS。
AMQP：
- AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。
JMS：
- MS即Java消息服务（JavaMessage Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。
AMQP 与 JMS 区别：
- JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
- JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
- JMS规定了两种消息模式（点对点、订阅）；而AMQP的消息模式更加丰富（最典型是路由）。
1.3. 消息队列产品
市场上常见的消息队列有如下：
- ActiveMQ：基于JMS
- ZeroMQ：基于C语言开发
- RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
- RocketMQ：基于JMS，阿里巴巴产品
- Kafka：类似MQ的产品；分布式消息系统，高吞吐量
1.4. RabbitMQ
RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。
- RabbitMQ官方地址：http://www.rabbitmq.com/
RabbitMQ提供了6种模式：简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式（远程调用，不太算MQ；不作介绍）；
2、安装及配置RabbitMQ
2.1. 安装说明
2.2. 用户以及Virtual Hosts配置
2.2.1、用户角色
RabbitMQ在安装好后，可以访问http://localhost:15672 ；其自带了guest/guest的用户名和密码；如果需要创建自定义用户，那么也可以登录管理界面后，如下操作：
角色说明：
1、 超级管理员(administrator)
- 可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。
2、 监控者(monitoring)
- 可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
3、 策略制定者(policymaker)
- 可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。
4、 普通管理者(management)
- 仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。
5、 其他
- 无法登陆管理控制台，通常就是普通的生产者和消费者。
2.2.2、Virtual Hosts配置
像mysql拥有数据库的概念并且可以指定用户对库和表等操作的权限。RabbitMQ也有类似的权限管理；在RabbitMQ中可以虚拟消息服务器Virtual Host，每个Virtual Hosts相当于一个相对独立的RabbitMQ服务器，每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。 相当于mysql的db。
注意：Virtual Name一般以/开头。
1. 创建Virtual Hosts
2. 设置Virtual Hosts权限
3、RabbitMQ案例入门
3.1、搭建rabbit模块工程及引入依赖
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
3.2、编写生产者
抽取创建connection的工具类com.ebuy.rabbitmq.util.ConnectionUtil工具类；
public class ConnectionUtil {
public static Connection getConnection() throws Exception
{
//创建连接工厂
ConnectionFactory connectionFactory=new ConnectionFactory();
//主机地址：默认为loalhost
connectionFactory.setHost("127.0.0.1");
//连接端口：默认为5672
connectionFactory.setPort(5672);
//虚拟主机名称：默认为 /
connectionFactory.setVirtualHost("/crm");
//连接用户名：新建的用户（默认为guest）
connectionFactory.setUsername("wbs");
//连接民密码：用户密码（默认为guest）
connectionFactory.setPassword("123456");
//新建连接
Connection connection=connectionFactory.newConnection();
return connection;
}
}
编写生产者：
public class Producer {
//队列名称
static final String QUEUE_NAME = "simple_queue";
public static void main(String[] args) {
try {
//创建一个连接
Connection connection = ConnectionUtil.getConnection();
//创建一个频道
Channel channel = connection.createChannel();
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(QUEUE_NAME,true,false,false,null);
//生产一个消息
String message="你好，我是001";
/**
* 参数1：交换机名称，如果没有指定则使用默认Default Exchage
* 参数2：路由key,简单模式可以传递队列名称
* 参数3：消息其它属性
* 参数4：消息内容(要转化为字节类型)
*/
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
//消息发送后，提示语句
System.out.println("消息已发送："+message);
//关闭连接
channel.close();
connection.close();
}catch (Exception e){
e.printStackTrace();
}
}
}
在执行上述的消息发送之后；可以登录rabbitMQ的管理控制台，可以发现队列和其消息：
3.3、编写消费者
编写消息的消费者com.ithouse.rabbitmq.simple.Consumer消费者类：
public class Consumer {
static final String QUEUE_NAME="simple_queue";
public static void main(String[] args) {
try {
//新建连接
Connection connection= ConnectionUtil.getConnection();
//创建频道
Channel channel=connection.createChannel();
/**
* 声明队列
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
*
* @param consumerTag 消息者标签，在channel.basicConsumer时候可以指定
* @param envelope 消息包的内容，可从中获取交换机，消息id，路由key，消息和重传标志(收到消息失败后是否需要重新发送)
* @param properties 属性信息
* @param body 消息
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("接收到的消息为：" + new String(body, "utf-8"));
}
};
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(QUEUE_NAME, true, consumer);
}catch (Exception e){
e.printStackTrace();
}
}
}
3.4、测试
先启动生产者：
再启动消费者：
3.5、小结
上述的入门案例中中其实使用的是如下的简单模式：
在上图的模型中，有以下概念：
P：生产者，也就是要发送消息的程序C：消费者：消息的接受者，会一直等待消息到来。queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
在rabbitMQ中消息者是一定要到某个消息队列中去获取消息的。
4、RabbitMQ工作模式
4.1、Work queues工作队列模式
Work Queues 与入门程序的 简单模式 相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。
4.1.1、测试work queue工作模式
Work Queues 与入门程序的 简单模式 的代码是几乎一样的；可以完全复制，并复制多一个消费者进行多个消费者同时消费消息的测试，只需在生产者中添加一个for循环模拟创建多个消息加入到消息队列中：
（1）、生产者
public class Producer {
static final String QUEUE_NAME="work_queue";
public static void main(String[] args) {
try {
//创建一个连接
Connection connection= ConnectionUtil.getConnection();
//创建一个频道
Channel channel=connection.createChannel();
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(QUEUE_NAME,true,false,false,null);
for(int i=1;i<=30;i++) {
String message = "你好，我是work："+i;
/**
* 参数1：交换机名称，如果没有指定则使用默认Default Exchage
* 参数2：路由key,简单模式可以传递队列名称
* 参数3：消息其它属性
* 参数4：消息内容
*/
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("消息已发送：" + message);
}
channel.close();
connection.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
（2）、消费者1
public class Consumer1 {
static final String QUEUE_NAME="work_queue";
public static void main(String[] args) {
try {
//新建连接
Connection connection= ConnectionUtil.getConnection();
//创建频道
Channel channel=connection.createChannel();
/**
* 声明队列
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
*
* @param consumerTag 消息者标签，在channel.basicConsumer时候可以指定
* @param envelope 消息包的内容，可从中获取交换机，消息id，路由key，消息和重传标志(收到消息失败后是否需要重新发送)
* @param properties 属性信息
* @param body 消息
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("Consumer1接收到的消息为：" + new String(body, "utf-8"));
/**
* 线程等待一秒
*/
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(QUEUE_NAME, true, consumer);
}catch (Exception e){
e.printStackTrace();
}
}
}
（3）、消费者2
public class Consumer2 {
static final String QUEUE_NAME="work_queue";
public static void main(String[] args) {
try {
//新建连接
Connection connection= ConnectionUtil.getConnection();
//创建频道
Channel channel=connection.createChannel();
/**
* 声明队列
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
*
* @param consumerTag 消息者标签，在channel.basicConsumer时候可以指定
* @param envelope 消息包的内容，可从中获取交换机，消息id，路由key，消息和重传标志(收到消息失败后是否需要重新发送)
* @param properties 属性信息
* @param body 消息
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("Consumer2接收到的消息为：" + new String(body, "utf-8"));
/**
* 线程等待一秒
*/
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(QUEUE_NAME, true, consumer);
}catch (Exception e){
e.printStackTrace();
}
}
}
4.1.2、测试
先启动两个消费者，然后再启动生产者发送消息；到两个消费者对应的控制台查看是否属于竞争性的接收到消息。
Consumer1：
Consumer2：
4.1.3、小结
在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。
4.2、订阅模式类型
订阅模式示例图：
前面2个案例中，只有3个角色：
P：生产者，也就是要发送消息的程序C：消费者：消息的接受者，会一直等待消息到来。queue：消息队列，图中红色部分
而在订阅模型中，多了一个Exchange角色，而且过程略有变化：
P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）C：消费者，消息的接受者，会一直等待消息到来。Queue：消息队列，接收消息、缓存消息。Exchange：交换机，就是图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：Fanout：广播，将消息交给所有绑定到交换机的队列；Direct：定向，把消息交给符合指定routing key的队列；Topic：通配符，把消息交给符合routing pattern（路由模式）的队列。
Exchange（交换机）只需负责转发消息，不具备储存消息的能力，因此如果没有任何队列与Exchange绑定或者没有符合路由规则的队列，那么消息会消失！！！
4.3、Publish/Subscribe发布与订阅模式
4.3.1、模式说明
发布订阅模式：
- 每个消费者监听自己的队列。
- 生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。
4.3.2、代码
（1）Consumer1
public class Consumer1 {
//交换机名称
static final String FANOUT_EXCHAGE = "fanout_exchange";
//队列名称1
static final String FANOUT_QUEUE_1 = "fanout_queue_1";
//队列名称1
static final String FANOUT_QUEUE_2 = "fanout_queue_2";
public static void main(String[] args) {
try {
//新建连接
Connection connection以上是关于RabbitMQ系列RabbitMQ入门的主要内容，如果未能解决你的问题，请参考以下文章