代码拉取完成,页面将自动刷新
typora-root-url | typora-copy-images-to |
---|---|
assert |
assert |
[TOC]
Java中有队列数据结构,但是是基于内存的,只有本JVM可以使用
MQ中间件是一个多客户端节点可以操作的队列结构
发送邮件、发送注册短信使用异步消息的模式,使得注册操作快速响应
扣减库存接口可能会根据库存系统的升级而升级,而不得不导致订单系统也需要升级(重新部署)
使用消息队列后,库存系统订阅队列即可,无需关心库存系统接口升级的问题
高并发场景下(秒杀),将请求存入mq,后台系统按照自己的处理能力来消费任务,达到削峰的目的
消息代理:指安装了消息中间件的服务器,用于接收消息和发送消息
通俗解释:消息代理接收到消息后会将消息继续发给目的地(生产者发送消息)
目的地主要有两种形式:队列、主题
点对点消息通信(point-to-point)
1.消息发送者发送消息,消息代理将其放入一个队列中,消息接受者从队列中获取消息内容,消息读取后被移出队列
2.队列可以被多个消费者监听,但一条消息只会被一个消费者成功消费
发布(publish)/订阅(subscribe)消息通信
1.发送者发送消息到主题,多个订阅者订阅该主题,多个消费者会同时收到消息
JMS:(Java Message Service)
JAVA消息服务,基于JVM信息代理的规范。ActiveMQ、HornetMQ是JMS实现
AMQP:(Advanced Message Queuing Protocol)
高级消息队列协议,也是一个消息代理的规范,兼容JMS
RabbitMQ是AMQP的实现
支持消息类型:byte[]=》只要能支持byte[]就可以传输,例如将对象转换为json,然后转二进制流传输即可
五种消息模型:重要,direct exchange其实就是队列
后四种就是主题的变种
使用@EnableRabbit开启自动配置
每条消息都需要在消息头中指定route-key
Exchange:交换器,每一个交换器都连接一个队列(可以看作局域网中的交换机的端口,每一个端口都连接一台电脑)
Queue队列:用于存储生产者发送的消息
交换器与Queue之间存在绑定关系,一个交换器可以绑定多个队列,存在多种绑定关系
根据消息头中的路由键指定的绑定关系发送到匹配的队列中
网络连接
生产者与Broker、消费者与Broker通过连接传输消息
一个客户端只会建立一条连接
一个客户端建立一条连接,一条连接内存储多个通道用于监听不同队列
只需要安装一个rabbitmq,但是可以分离出多个微主机,互相之间配置隔离,使用不同的url访问
1.例如dev、test可以使用不同VHost
2.不同的系统使用不同的VHost
简介:
1.可靠抵达:消息可靠抵达MQ,消息可靠抵达消费者
2.事务消息:性能下降250倍,所以采用确认机制代替
文档:https://www.rabbitmq.com/confirms.html#publisher-confirms
Reliable Delivery=》Acknowledgements and Confirms=>Publisher confirms
简介:
1.生产者发送消息到Queue会经过两个两个过程【确认机制看做一种协议】
1)消息从publisher到达Broker(到达后会回调confirmCallback,消费者被告知消息是否抵达服务器)
【集群情况下必须所有的broker接收到才会调用confirmCallback】
2)消息从Exchange投递到Queue(失败后会回调returnCallback,消费者被告知消息是否抵达Queue)
# 开启发送端确认:
spring.rabbitmq.publisher-returns=true
# 消息在没有被队列接收时是否强行退回
spring.rabbitmq.template.mandatory=true
// 测试步骤,调用单元测试中发送消息的方法,触发回调
@Configuration
public class MyRabbitConfig {
@Autowired
RabbitTemplate rabbitTemplate;
@Bean
public MessageConverter messageConverter() {
// 使用json序列化器来序列化消息,发送消息时,消息对象会被序列化成json格式
return new Jackson2JsonMessageConverter();
}
/**
* 定制RabbitTemplate
* 1、服务收到消息就会回调
* 1、spring.rabbitmq.publisher-confirms: true
* 2、设置确认回调
* 2、消息正确抵达队列就会进行回调
* 1、spring.rabbitmq.publisher-returns: true
* spring.rabbitmq.template.mandatory: true
* 2、设置确认回调ReturnCallback
* <p>
* 3、消费端确认(保证每个消息都被正确消费,此时才可以broker删除这个消息)
*/
@PostConstruct // (MyRabbitConfig对象创建完成以后,执行这个方法)
public void initRabbitTemplate() {
/**
* 发送消息触发confirmCallback回调
* @param correlationData:当前消息的唯一关联数据(如果发送消息时未指定此值,则回调时返回null)
* @param ack:消息是否成功收到(ack=true,消息抵达Broker)
* @param cause:失败的原因
*/
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
System.out.println("发送消息触发confirmCallback回调" +
"\ncorrelationData ===> " + correlationData +
"\nack ===> " + ack + "" +
"\ncause ===> " + cause);
});
}
}
简介:
1.生产者发送消息到Queue会经过两个两个过程【确认机制看做一种协议】
1)消息从publisher到达Broker(到达后会回调confirmCallback,消费者被告知消息是否抵达服务器)
2)消息从Exchange投递到Queue(失败后会回调returnCallback,消费者被告知消息是否抵达Queue)
# 开启发送端抵达队列确认
spring.rabbitmq.publisher-returns=true
@Configuration
public class MyRabbitConfig {
@Autowired
RabbitTemplate rabbitTemplate;
@Bean
public MessageConverter messageConverter() {
// 使用json序列化器来序列化消息,发送消息时,消息对象会被序列化成json格式
return new Jackson2JsonMessageConverter();
}
/**
* 定制RabbitTemplate
* 1、服务收到消息就会回调
* 1、spring.rabbitmq.publisher-confirms: true
* 2、设置确认回调
* 2、消息正确抵达队列就会进行回调
* 1、spring.rabbitmq.publisher-returns: true
* spring.rabbitmq.template.mandatory: true
* 2、设置确认回调ReturnCallback
* <p>
* 3、消费端确认(保证每个消息都被正确消费,此时才可以broker删除这个消息)
*/
@PostConstruct // (MyRabbitConfig对象创建完成以后,执行这个方法)
public void initRabbitTemplate() {
/**
* 消息未到达队列触发returnCallback回调
* 只要消息没有投递给指定的队列,就触发这个失败回调
* @param message:投递失败的消息详细信息
* @param replyCode:回复的状态码
* @param replyText:回复的文本内容
* @param exchange:接收消息的交换机
* @param routingKey:接收消息的路由键
*/
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
// 需要修改数据库 消息的状态【后期定期重发消息】
System.out.println("消息未到达队列触发returnCallback回调" +
"\nmessage ===> " + message +
"\nreplyCode ===> " + replyCode +
"\nreplyText ===> " + replyText +
"\nexchange ===> " + exchange +
"\nroutingKey ===> " + routingKey);
});
}
}
简介:
1.消费者接收消息也会经过ack消息确认机制,只有消费者成功接收消息,broker才允许删除消息
2.默认情况下消息抵达客户端后自动确认,服务端消息自动删除
3.手动确认模式下,消费者接收消息后但是不执行ack/nack进行确认,服务端队列中的消息会从unacked状态变为ready状态等待下一次消费(即使consumer宕机消息也不会丢失)
4.注意:消息到达消费端,消息会进入unacked状态,如果手动确认模式下,消费者因宕机而未执行ack/nack,消息会一直处于unacked状态不会被删除(持久化了)
5.签收,ack
# 消费者手动确认模式,关闭自动确认,否则会消息丢失
spring.rabbitmq.listener.simple.acknowledge-mode=manual
@RabbitListener(queues = {"hello-java-queue"})
@Service("orderItemService")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
@RabbitHandler
public void revieveMessage(Message message, OrderEntity entity, Channel channel) {
// 请求体,序列化存储(本例中已使用Jackson2JsonMessageConverter序列化器作JSON序列化存储)
byte[] body = message.getBody();
// 请求头
MessageProperties properties = message.getMessageProperties();
// channel内按顺序自增的long类型消息标签
long deliveryTag = properties.getDeliveryTag();
// JSON反序列得到消息内容对象
OrderEntity reason = JSONObject.parseObject(body, OrderEntity.class);
System.out.println("接受到的消息对象" + message);
System.out.println("接受到的消息内容" + reason);
System.out.println("接受到的消息内容" + entity);
try {
if (deliveryTag == 2) {
// 手动确认,消息会从unacked中删除,total数量减1
// boolean multiple:是否批量签收
channel.basicAck(deliveryTag, false);
} else {
// 手动拒签
// boolean multiple:是否批量拒签
// boolean requeue:当前拒签消息是否发回服务器重新入队
channel.basicNack(deliveryTag, false, true);
}
} catch (IOException e) {
// 网络中断
e.printStackTrace();
}
}
}
场景:
未付款订单,超时后,系统自动取消订单并释放占有物品
常用解决方案:
Spring的Schedule定时任务轮询数据库
缺点:
消耗系统内存、增加数据库的压力、存在较大的时间误差
解决:
在保证可靠消息的前提下,使用延时队列+死信队列,达到最终一致性(柔性事务)
TTL(Time To Live)
死信Exchange其实就是普通的Exchange
队列设置好自己的Dead Letter Exchange,当此队列中的消息过期后会被转发到这个路由,被称为死信路由
TTL消息 + 死信Exchange
使用一个队列接收死信Exchange中的TTL消息,这样的队列被称为延时队列
注意:存放TTL消息的队列不要让客户端监听(这个队列和延时队列不是同一个,延时队列是存储已经超时的TTL消息)
实现1:给队列设置TTL
arguments.put("x-dead-letter-exchange", "order-event-exchange");// 死信路由
arguments.put("x-dead-letter-routing-key", "order.release.order");// 死信路由键
arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟
场景:
配合延时队列+死信队列实现最终一致性(柔性事务)
CREATE TABLE `mq_message` (
`message_id` char(32) NOT NULL,
`content` json,
`to_exchane` varchar(255) DEFAULT NULL,
`routing_key` varchar(255) DEFAULT NULL,
`class_type` varchar(255) DEFAULT NULL,
`message_status` int(1) DEFAULT '0' COMMENT '0-新建 1-已发送 2-错误抵达 3-已抵达',
`create_time` datetime DEFAULT NULL,
`update_time` datetime DEFAULT NULL,
PRIMARY KEY (`message_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
前提:
使用定时器扫描mq_message定时重发
情况1:网络连接失败,消息未抵达Broker
解决:发送消息时同时将消息持久化到MQ中并设定状态为已抵达
当出现异常时在catch处修改消息状态为错误抵达
情况2:消息抵达Broker,但为抵达queue,消息会丢失(只有抵达了queue消息才会持久化)
解决:开启生产者确认机制,将触发returnCallback的returnedMessage的消息状态修改为错误抵达
情况3:消费者未ack时宕机,导致消息丢失
解决:开启消费者手动ack
情况1:业务逻辑已经执行,但是ack时宕机,消息由unack变为ready,消息重新入队
解决:将接口设计成幂等性,例如库存解锁时判断工作单的状态,已解锁则无操作
解决2:防重表
情况1:生产者流量太大
解决:减慢发送消息速率(验证码、防刷、重定向、削峰)
情况2:消费者能力不足或宕机
解决:上线更多消费者
解决2:上线专门的队列消费服务,批量取出消息入库,离线处理业务慢慢处理
可以添加一个消息服务,各模块调用发送消息API即可
实现消息存库+异常修改状态
思考:如果feign调用失败没有问题,做好本地事务,feign调用失败回滚即可
场景:
高并发秒杀模块,将秒杀成功创建订单的消息发送给MQ,订单模块按照自己的能力消费生成订单
这个过程就是队列削峰(不走购物车逻辑,否则秒杀的高并发流量会带给订单模块)
文档:https://www.rabbitmq.com/networking.html
4369,25672(Erlang发现&集群端口)
5672,5671(AMQP端口)
15672(web管理后台端口)
61613,61614(STOMP协议端口)
1883,8883(MQTT协议端口)
https://www.rabbitmq.com/networking.html
1.下载镜像并启动
docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management
2.修改为自动重启
docker update rabbitmq --restart=always
3.登录管理页面
http://192.168.56.10:15672/
guest
guest
1.新增了Exchange和Binding角色
2.消息路由:消息到达Exchange,并根据Binding关系发布到Queue队列中的过程,被称为消息路由
1.Exchange名词解释:是AMQP高级消息队列协议的划分
2.RabbitMQ有四种Exchange类型:
2.1.点对点通信
2.1.1.direct
解析:直接交换机【单播模式,按照路由关系精确匹配】
例:路由键是dog,direct Exchange绑定了两个队列,队列1:dog、队列2:dogTom
使用direct Exchange只会把消息发送给队列1
2.1.2.headers【性能比较低,不会用到,与direct几乎一致】
2.2.发布/订阅:
2.2.1.fanout
解析:扇形【广播模式,不关心路由键】
例:消息会发送给fanout Exchange绑定的所有队列
2.2.2.topic
解析:主题【部分广播,区分路由键】
例:binding.key=usa.* :*处必须有一个单词
binding.key=#.news:#处有0个或多个单词
注:不能匹配字母,只能匹配单词
3.
direct案例:
fanout案例:
Topic案例:
http://192.168.56.10:15672/#/exchanges
1.新建交换机
Exchanges=》Add a new Exchange=》Add Exchange
2.新建队列
Queues=》Add a new Queue=》Add Queue
3.在交换机表格中点击新建的交换机,绑定刚刚新建的队列
Bindings=》Add binding from this exchange=》To queue=》Bind
4.发送消息
=》publishe Message
5.获取消息
Nack message requeue true
Automatic ack
Reject requeue true
Reject requeue false
6.解除binding
=》unbind
消息发送成功:
不回复消息,并且消息重新入队
自动回复消息,消息不重新入队
根据上表
1.建立交换机:
exchange.direct【点对点】
exchange.fanout【广播,消息会发送给binding的多个队列中】
exchange.topic【广播,会先找到存在binding关系的队列,然后按照binding关系的路由规则与路由键进行模糊匹配】
2.建立队列:
atguigu
atguigu.news
atguigu.emps
gulixueyuan.news
3.创建binding关系
exchange.direct绑定所有队列【绑定规则是根据路由键与路由精确绑定决定消息进入哪个队列】
exchange.fanout绑定所有队列【消息会发送给每个队列】
exchange.topic绑定所有队列【并设置不同的路由键atguigu.#、*.news】
4.发送消息
1.订单服务中引入依赖,场景启动器,引入后RabbitAutoConfiguration自动生效
<!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2.RabbitAutoConfiguration生效后自动注入多个容器:
CachingConnectionFactory:
RabbitTemplate:
AmqpAdmin:
RabbitMessagingTemplate:
3.RabbitProperties:配置类
用于创建Exchange、Queue、Binding
启动类标注:(只有需要用到监听消息时才需要该注解,开启后可以使用@RabbitListener)
@EnableRabbit
spring:
rabbitmq:
host: 192.168.56.10
port: 5672
# 虚拟主机
virtual-host: /
# 开启发送端抵达队列确认【发送端确认机制+本地事务表】
publisher-returns: true
# 开启发送确认【发送端确认机制+本地事务表】
publisher-confirm-type: correlated
# 只要抵达队列,优先回调return confirm
template:
mandatory: true
# 使用手动确认模式,关闭自动确认【消息丢失】
listener:
simple:
acknowledge-mode: manual
@Configuration
public class MyRabbitConfig {
@Bean
public MessageConverter messageConverter() {
// 使用json序列化器来序列化消息,发送消息时,消息对象会被序列化成json格式
return new Jackson2JsonMessageConverter();
}
}
@Autowired
AmqpAdmin amqpAdmin;
@Test
void createExchange() {
// 创建交换机
// String name, boolean durable, boolean autoDelete
DirectExchange exchange = new DirectExchange("hello-java-exchange", true, false);
amqpAdmin.declareExchange(exchange);
log.info("Exchange创建[{}]成功", "hello-java-exchange");
}
@Autowired
AmqpAdmin amqpAdmin;
@Test
void createQueue() {
// 创建队列
// String name, boolean durable, boolean exclusive, boolean autoDelete
// exclusive:是否排他,true:只有一个连接可以使用此队列,其他连接无法连上此队列
Queue queue = new Queue("hello-java-queue", true, false, false);
amqpAdmin.declareQueue(queue);
log.info("Queue创建[{}]成功", "hello-java-queue");
}
@Autowired
AmqpAdmin amqpAdmin;
@Test
void createBinding() {
// 创建绑定,交换机绑定目的地
// String destination:目的地name
// DestinationType destinationType:目的地类型【queue或exchange(路由)】
// String exchange:待绑定交换机
// String routingKey:路由键
Binding bind = new Binding("hello-java-queue", Binding.DestinationType.QUEUE,"hello-java-exchange", "hello.java", null)
amqpAdmin.declareBinding(bind);
log.info("Binding创建[{}]成功", "hello-java-binding");
}
@Autowired
RabbitTemplate rabbitTemplate;
@Test
void sendMsg() {
// 如果发送的消息是个对象,会使用序列化机制,将对象写出去。对象类必须实现 serializable
// 如果使用JSON序列化器,则不需要类实现Serializable
// String exchange:交换机
// String routingKey:路由键
// final Object object:消息
// CorrelationData correlationData:可指定消息ID
// 消息对象,可以是任意类型,类必须实现serializable,消息会以序列化的方式写入流中
OrderReturnReasonEntity message = new OrderReturnReasonEntity();// 退货原因
message.setId(1L);
message.setCreateTime(new Date());
message.setName("哈哈");
// 消息ID
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", message, correlationData);
}
简介:
1.用于标注在监听类或监听方法上,接收消息,需要指定监听的队列(数组)
2.使用该注解之前,需要在启动类加上该注解:@EnableRabbit
3.@RabbitListener即可以标注在方法上又可以标注在类上
标注在类上:表示该类是监听类,使得@RabbitHandler注解生效
标注在方法上:表示该方法时监听方法,会监听指定队列获得消息
4.一般只标注在方法上,并配合@RabbitHandler使用,重载的方式接收不同消息对象
import com.rabbitmq.client.Channel;
@Service("orderItemService")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
/**
* queues:声明需要监听的队列
* channel:当前传输数据的通道
* 获取实际消息内容有两种方式:
* 方式一:在方法参数列表中直接声明出来
* 方式二:从请求体中取出消息的二进制形式,然后通过JSON反序列化即可
*/
@RabbitListener(queues = {"hello-java-queue"})
public void revieveMessage(Message message, OrderReturnReasonEntity entity, Channel channel) {
// 请求体,序列化存储(本例中已使用Jackson2JsonMessageConverter序列化器作JSON序列化存储)
byte[] body = message.getBody();
// 请求头
MessageProperties messageProperties = message.getMessageProperties();
// JSON反序列得到消息内容对象
OrderReturnReasonEntity reason = JSONObject.parseObject(body, OrderReturnReasonEntity.class);
System.out.println("接受到的消息对象" + message);
System.out.println("接受到的消息内容" + reason);
System.out.println("接受到的消息内容" + entity);
}
}
简介:
复制一份订单配置9011
-Xmx100m
--server.port=9011
同时运行多个订单模块,然后测试发送多条消息
结论:
1.多个客户端可以共同监听同一队列
2.一条消息同时只能被一个客户端接收
3.同一个客户端接收消息是串行的,revieveMessage方法执行完后才会继续接收下一条消息
9010接收消息:
9011接收消息:
作用:
配合@RabbitListener,使用方法重载的方法接收不同的消息类型
简介:
1.用于标注在监听方法上,接收消息,不需要指定监听的队列
2.使用该注解之前,需要在启动类加上该注解:@EnableRabbit
3.@RabbitListener只可以标注在方法,重载的方式接收不同消息对象
@Slf4j
@SpringBootTest
class GulimallOrderApplicationTests {
@Autowired
RabbitTemplate rabbitTemplate;
@Test
void sendMsg2() {
// 该测试案例向同一队列发送了两个不同类型的消息对象
// 消息一:
OrderReturnReasonEntity message1 = new OrderReturnReasonEntity();// 退货原因
message1.setId(1L);
message1.setCreateTime(new Date());
message1.setName("哈哈");
CorrelationData correlationData1 = new CorrelationData(UUID.randomUUID().toString());// 消息ID
rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", message1, correlationData1);
// 消息二:
OrderEntity message2 = new OrderEntity();// 退货原因
message2.setId(1L);
message2.setCreateTime(new Date());
message2.setOrderSn("哈哈");
CorrelationData correlationData2 = new CorrelationData(UUID.randomUUID().toString());// 消息ID
rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", message2, correlationData2);
}
}
@RabbitListener(queues = {"hello-java-queue"})
@Service("orderItemService")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
/**
* 方法重载1
*/
@RabbitHandler
public void revieveMessage(Message message, OrderReturnReasonEntity entity, Channel channel) {
// 请求体,序列化存储(本例中已使用Jackson2JsonMessageConverter序列化器作JSON序列化存储)
byte[] body = message.getBody();
// 请求头
MessageProperties messageProperties = message.getMessageProperties();
// JSON反序列得到消息内容对象
OrderReturnReasonEntity reason = JSONObject.parseObject(body, OrderReturnReasonEntity.class);
System.out.println("接受到的消息对象" + message);
System.out.println("接受到的消息内容" + reason);
System.out.println("接受到的消息内容" + entity);
}
/**
* 方法重载2
*/
@RabbitHandler
public void revieveMessage(Message message, OrderEntity entity, Channel channel) {
// 请求体,序列化存储(本例中已使用Jackson2JsonMessageConverter序列化器作JSON序列化存储)
byte[] body = message.getBody();
// 请求头
MessageProperties messageProperties = message.getMessageProperties();
// JSON反序列得到消息内容对象
OrderEntity reason = JSONObject.parseObject(body, OrderEntity.class);
System.out.println("接受到的消息对象" + message);
System.out.println("接受到的消息内容" + reason);
System.out.println("接受到的消息内容" + entity);
}
}
spring:
rabbitmq:
host: 192.168.56.10
port: 5672
# 虚拟主机
virtual-host: /
# 开启发送端发送确认,无论是否到达broker都会触发回调【发送端确认机制+本地事务表】
publisher-confirm-type: correlated
# 开启发送端抵达队列确认,消息未被队列接收时触发回调【发送端确认机制+本地事务表】
publisher-returns: true
# 消息在没有被队列接收时是否强行退回
template:
mandatory: true
# 消费者手动确认模式,关闭自动确认,否则会消息丢失
listener:
simple:
acknowledge-mode: manual
描述:
可以共用一个exchange,指定不同的路由分别绑定延时队列和死信队列
/**
* 创建队列,交换机,延时队列,绑定关系 的configuration
* 1.Broker中的Queue、Exchange、Binding不存在的情况下,会自动创建(在RabbitMQ),不会重复创建覆盖
* 2.懒加载,只有第一次使用的时候才会创建(例如监听队列)
*/
@Configuration
public class MyRabbitMQConfig {
/**
* 延时队列
*/
@Bean
public Queue orderDelayQueue() {
/**
* Queue(String name, 队列名字
* boolean durable, 是否持久化
* boolean exclusive, 是否排他
* boolean autoDelete, 是否自动删除
* Map<String, Object> arguments) 属性【TTL、死信路由、死信路由键】
*/
HashMap<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", "order-event-exchange");// 死信路由
arguments.put("x-dead-letter-routing-key", "order.release.order");// 死信路由键
arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟
return new Queue("order.delay.queue", true, false, false, arguments);
}
/**
* 交换机(死信路由)
*/
@Bean
public Exchange orderEventExchange() {
return new TopicExchange("order-event-exchange", true, false);
}
/**
* 死信队列
*/
@Bean
public Queue orderReleaseQueue() {
return new Queue("order.release.order.queue", true, false, false);
}
/**
* 绑定:交换机与订单解锁延迟队列
*/
@Bean
public Binding orderCreateBinding() {
/**
* String destination, 目的地(队列名或者交换机名字)
* DestinationType destinationType, 目的地类型(Queue、Exhcange)
* String exchange,
* String routingKey,
* Map<String, Object> arguments
**/
return new Binding("order.delay.queue",
Binding.DestinationType.QUEUE,
"order-event-exchange",
"order.create.order",
null);
}
/**
* 绑定:交换机与订单解锁死信队列
*/
@Bean
public Binding orderReleaseBinding() {
return new Binding("order.release.order.queue",
Binding.DestinationType.QUEUE,
"order-event-exchange",
"order.release.order",
null);
}
/**
* 绑定:交换机与库存解锁
*/
@Bean
public Binding orderReleaseOtherBinding() {
return new Binding("stock.release.stock.queue",
Binding.DestinationType.QUEUE,
"order-event-exchange",
"order.release.other.#",
null);
}
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。