文章目录
目录
文章目录
前言
一、简单介绍
二、使用步骤
2.1 环境配置
2.2 测试使用--java
2.2.1 简单模式
2.2.2 Work queues 工作队列模式
2.2.3 Pub/Sub 订阅模式
2.2.4 Routing 路由模式
总结
前言
MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。
MQ，消息队列，存储消息的中间件
分布式系统通信两种方式：直接远程调用 和 借助第三方 完成间接通信
发送方称为生产者，接收方称为消费者
MQ 的优势和劣势
优势
应用解耦：提高系统容错性和可维护性
异步提速：提升用户体验和系统吞吐量
削峰填谷：提高系统稳定性
劣势
系统可用性降低：系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用
系统复杂度提高：MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性
一致性问题：A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理失败。如何保证消息数据处理的一致性？
一、简单介绍
既然 MQ 有优势也有劣势，那么使用 MQ 需要满足什么条件呢？
生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。
容许短暂的不一致性。
确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本。
目前业界有很多的 MQ 产品，例如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等，也有直接使用 Redis 充当消息队列的案例，而这些消息队列产品，各有侧重，在实际选型时，需要结合自身需求及 MQ 产品特征，综合考虑
AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。2006年，AMQP 规范发布。类比HTTP。
RabbitMQ 中的相关概念：
Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker
Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
Connection：publisher／consumer 和 broker 之间的 TCP 连接. 长连接
Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销
Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)
Queue：消息最终被送到这里等待 consumer 取走. 存储在队列中
Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据
二、使用步骤
2.1 环境配置
在一台安装有docker上的虚拟机上下载RabbitMQ镜像
执行以下命令
docker run -d --name rabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 -p 25672:25672 -p 61613:61613 -p 1883:1883 rabbitmq:management
该命令会自动拉取镜像 并后台启动该容器
然后在浏览器上输入http://虚拟机地址ip:15672/#/ 就能访问rabbitmq的可视化界面
2.2 测试使用--java
新建两个maven工程 一个是生产者 一个是消费者
每个里面引入依赖
<!--rabbitmq连接依赖--> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.16.0</version> </dependency>
2.2.1 简单模式
生产者端代码
/**
* 测试简单模式
* @throws Exception
*/
@Test
public void testSimple() throws Exception{
//创建连接工厂对象---配置连接信息
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
factory.setHost("192.168.223.158");
factory.setPort(5672);
//创建连接对象Connection
Connection connection = factory.newConnection();
//创建channel信道
Channel channel=connection.createChannel();
//创建队列
/**
* String queue,队列名称
* boolean durable, 是否持久化
* boolean exclusive,是否私有
* boolean autoDelete,是否自动删除
* Map<String, Object> arguments: 队列的参数
*/
channel.queueDeclare("simple_queue",true,false,false,null);
//发消息
/**
* String exchange, 简单模式没有交换机所以写为""
* String routingKey, 路由key-- 默认队列名
* BasicProperties props, 消息的属性
* byte[] body: 消息的内容
*/
String msg="想睡觉吗? 那就睡吧";
channel.basicPublish("","simple_queue",null,msg.getBytes());
channel.close();
connection.close();
}消费者端代码
public class TestSimple {
public static void main(String[] args) throws Exception {
//创建连接工厂对象---配置连接信息
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
factory.setHost("192.168.223.158");
factory.setPort(5672);
//创建连接对象Connection
Connection connection = factory.newConnection();
//创建channel信道
Channel channel=connection.createChannel();
//创建队列:如果存在指定的队列名 则不创建
/**
* String queue,队列名称
* boolean durable, 是否持久化
* boolean exclusive,是否私有
* boolean autoDelete,是否自动删除
* Map<String, Object> arguments: 队列的参数
*/
channel.queueDeclare("simple_queue",true,false,false,null);
//获取消息
/**
* (String queue,队列名
* boolean autoAck, 自动确认
* Consumer callback : 回调函数.
*/
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//body就是从队列中获取的消息内容
String msg = new String(body);
System.out.println("接受的消息内容:"+msg);
//根据消息的内容写自己的业务代码
}
};
channel.basicConsume("simple_queue",true,consumer);
}
}
简单模式 生产者发送消息到队列。然后消费者直接就获取该消息处理掉了
2.2.2 Work queues 工作队列模式
在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。
Work Queues 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如：短信服务部署多个，只需要有一个节点成功发送即可。
生产者代码
/**
* 工作模式
* @throws Exception
*/
@Test
public void testWorker() throws Exception{
//创建连接工厂对象---配置连接信息
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
factory.setHost("192.168.223.158");
factory.setPort(5672);
//创建连接对象Connection
Connection connection = factory.newConnection();
//创建channel信道
Channel channel=connection.createChannel();
//创建队列
/**
* String queue,队列名称
* boolean durable, 是否持久化
* boolean exclusive,是否私有
* boolean autoDelete,是否自动删除
* Map<String, Object> arguments: 队列的参数
*/
channel.queueDeclare("worker_queue",true,false,false,null);
//发消息
/**
* String exchange, 简单模式没有交换机所以写为""
* String routingKey, 路由key-- 默认队列名
* BasicProperties props, 消息的属性
* byte[] body: 消息的内容
*/
for(int i=0;i<20;i++) {
String msg = "想睡觉吗? 忍忍就过去了----->"+i;
channel.basicPublish("", "worker_queue", null, msg.getBytes());
}
channel.close();
connection.close();
}
两个消费者代码
public class TestWorker01 {
public static void main(String[] args) throws Exception {
//创建连接工厂对象---配置连接信息
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
factory.setHost("192.168.223.158");
factory.setPort(5672);
//创建连接对象Connection
Connection connection = factory.newConnection();
//创建channel信道
final Channel channel=connection.createChannel();
//每次处理一条消息
channel.basicQos(1);
//创建队列:如果存在指定的队列名 则不创建
/**
* String queue,队列名称
* boolean durable, 是否持久化
* boolean exclusive,是否私有
* boolean autoDelete,是否自动删除
* Map<String, Object> arguments: 队列的参数
*/
channel.queueDeclare("worker_queue",true,false,false,null);
//获取消息
/**
* (String queue,队列名
* boolean autoAck, 自动确认
* Consumer callback : 回调函数.
*/
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
//body就是从队列中获取的消息内容
String msg = new String(body);
System.out.println("接受的消息内容:"+msg);
//根据消息的内容写自己的业务代码
//long deliveryTag, boolean multiple
//手动告诉队列消息处理完毕
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
channel.basicConsume("worker_queue",false,consumer);
}
}
public class TestWorker02 {
public static void main(String[] args) throws Exception {
//创建连接工厂对象---配置连接信息
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
factory.setHost("192.168.223.158");
factory.setPort(5672);
//创建连接对象Connection
Connection connection = factory.newConnection();
//创建channel信道
final Channel channel=connection.createChannel();
//每次处理一条消息
channel.basicQos(1);
//创建队列:如果存在指定的队列名 则不创建
/**
* String queue,队列名称
* boolean durable, 是否持久化
* boolean exclusive,是否私有
* boolean autoDelete,是否自动删除
* Map<String, Object> arguments: 队列的参数
*/
channel.queueDeclare("worker_queue",true,false,false,null);
//获取消息
/**
* (String queue,队列名
* boolean autoAck, 自动确认
* Consumer callback : 回调函数.
*/
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//body就是从队列中获取的消息内容
String msg = new String(body);
System.out.println("接受的消息内容:"+msg);
//根据消息的内容写自己的业务代码
//long deliveryTag, boolean multiple
//手动告诉队列消息处理完毕
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
channel.basicConsume("worker_queue",false,consumer);
}
}
如果其中一个消费者处理时间慢的话，那么另一个会把所有的消息都消费完。
2.2.3 Pub/Sub 订阅模式
在订阅模型中，多了一个 Exchange 角色，而且过程略有变化：
P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
C：消费者，消息的接收者，会一直等待消息到来
Queue：消息队列，接收消息、缓存消息
Exchange：交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：
Fanout：广播，将消息交给所有绑定到交换机的队列
Direct：定向，把消息交给符合指定routing key 的队列
Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！
生产者代码
/**
* 发布订阅模式
* @throws Exception
*/
@Test
public void testPublisher() throws Exception{
//创建连接工厂对象---配置连接信息
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
factory.setHost("192.168.223.158");
factory.setPort(5672);
//创建连接对象Connection
Connection connection = factory.newConnection();
//创建channel信道
Channel channel=connection.createChannel();
//创建交换机
/**
* String exchange, 交换机的名称
* BuiltinExchangeType type, 交换机的类型
* boolean durable: 是否持久化
*/
channel.exchangeDeclare("fanout_exchange",BuiltinExchangeType.FANOUT,true);
//创建队列
channel.queueDeclare("fanout_queue01",true,false,false,null);
channel.queueDeclare("fanout_queue02",true,false,false,null);
//队列交换机绑定
channel.queueBind("fanout_queue01","fanout_exchange","");
channel.queueBind("fanout_queue02","fanout_exchange","");
String msg="想睡觉吗? 忍忍就过去了????";
channel.basicPublish("fanout_exchange","",null,msg.getBytes());
}
消费者的代码
public class TestPublisher01 {
public static void main(String[] args) throws Exception {
//创建连接工厂对象---配置连接信息
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
factory.setHost("192.168.223.158");
factory.setPort(5672);
//创建连接对象Connection
Connection connection = factory.newConnection();
//创建channel信道
final Channel channel=connection.createChannel();
//每次处理一条消息
channel.basicQos(1);
//获取消息
/**
* (String queue,队列名
* boolean autoAck, 自动确认
* Consumer callback : 回调函数.
*/
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
//body就是从队列中获取的消息内容
String msg = new String(body);
System.out.println("接受的消息内容:"+msg);
//根据消息的内容写自己的业务代码
//long deliveryTag, boolean multiple
//手动告诉队列消息处理完毕
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
channel.basicConsume("fanout_queue01",false,consumer);
}
}public class TestPublisher02 {
public static void main(String[] args) throws Exception {
//创建连接工厂对象---配置连接信息
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
factory.setHost("192.168.223.158");
factory.setPort(5672);
//创建连接对象Connection
Connection connection = factory.newConnection();
//创建channel信道
final Channel channel=connection.createChannel();
//每次处理一条消息
channel.basicQos(1);
//获取消息
/**
* (String queue,队列名
* boolean autoAck, 自动确认
* Consumer callback : 回调函数.
*/
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
//body就是从队列中获取的消息内容
String msg = new String(body);
System.out.println("接受的消息内容:"+msg);
//根据消息的内容写自己的业务代码
//long deliveryTag, boolean multiple
//手动告诉队列消息处理完毕
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
channel.basicConsume("fanout_queue02",false,consumer);
}
}
2.2.4 Routing 路由模式
队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）
消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey
Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的Routingkey 与消息的 Routing key 完全一致，才会接收到消息
P：生产者，向 Exchange 发送消息，发送消息时，会指定一个routing key
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列
C1：消费者，其所在队列指定了需要 routing key 为 error 的消息
C2：消费者，其所在队列指定了需要 routing key 为 info、error、warning 的消息
生产者代码
@Test
public void testRouter() throws Exception{
//创建连接工厂对象---配置连接信息
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("admin");
factory.setPassword("admin");
factory.setVirtualHost("/");
factory.setHost("192.168.223.158");
factory.setPort(5672);
//创建连接对象Connection
Connection connection = factory.newConnection();
//创建channel信道
Channel channel=connection.createChannel();
//创建交换机
/**
* String exchange, 交换机的名称
* BuiltinExchangeType type, 交换机的类型
* boolean durable: 是否持久化
*/
channel.exchangeDeclare("router_exchange",BuiltinExchangeType.DIRECT,true);
//创建队列
channel.queueDeclare("router_queue01",true,false,false,null);
channel.queueDeclare("router_queue02",true,false,false,null);
//队列交换机绑定
channel.queueBind("router_queue01","router_exchange","error");
channel.queueBind("router_queue02","router_exchange","error");
channel.queueBind("router_queue02","router_exchange","info");
channel.queueBind("router_queue02","router_exchange","warning");
String msg="这都是你的警告---warning";
channel.basicPublish("router_exchange","warning",null,msg.getBytes());
}
Routing 模式要求队列在绑定交换机时要指定 routing key，消息会转发到符合 routing key 的队列。
总结
待补充
