网站网站到底怎么做,重庆广告片制作,大庆油田内网主页网址,附近最好的装修公司二、高级特性、应用问题以及集群搭建
高级特性
1.消息的可靠性投递
在使用RabbitMQ的时候#xff0c;作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。 rabbitMQ整个消息投递的路径为#xff1a; produ…二、高级特性、应用问题以及集群搭建
高级特性
1.消息的可靠性投递
在使用RabbitMQ的时候作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。 rabbitMQ整个消息投递的路径为 producer - rabbitMQ broker - exchange - queue - consumer
confirm确认模式 confirm确认模式是再producer传递给exchange过程中控制消息的模式当消息成功的从producer传递到了exchange那么则会返回一个 confirmCallBack() 回调函数return 退回模式 return退回模式是指消息从exchange传递给queue过程中消息传递失败则会返回一个returnCallBack() 回调函数
1.1 confirm确认模式的代码编写
因为确认模式是producer到exchange所以代码和配置修改应该写在生产者的模块中。 第一步开启确认模式 新版本的rabbitmq弃用了publish-confirmstrue可以改用 publisher-confirm-type: correlated实现同样的效果 spring:rabbitmq:password: heimausername: heimaport: 5673virtual-host: itcasthost: 1.12.244.105#开启确认模式publisher-confirm-type: correlated第二步编写confirmCallBack()函数 回调函数confirm()的返回值在发送消息成功时ack为true但是我遇到一个问题就是消息发送成功了在队列中也能看到但是返回值ack为false clean channel shutdown; 这是因为convertAncSend()方法结束后rabbitMQ的资源也就关闭了所以就算成功了回调函数返回值也是false所以我们在后面强制睡眠200ms让资源晚点关闭这样的话得到的ack就是true了
package com.rabbitmq.springboot_mqproducer;import com.rabbitmq.springboot_mqproducer.rabbitMQconfig.MQConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;import javax.annotation.Resource;SpringBootTest
class SpringbootMqProducerApplicationTests {ResourceRabbitTemplate rabbitTemplate;Testvoid contextLoads() throws InterruptedException {rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/**** param correlationData 相关的配置信息* param b 消息是否发送成功* param s 消息发送失败原因*/Overridepublic void confirm(CorrelationData correlationData, boolean b, String s) {System.out.println(confirm方法被执行了);System.out.println(b);if(b){System.out.println(消息从producer - exchange成功);System.out.println(失败原因 s);}else{System.out.println(消息从producer - exchange失败);System.out.println(失败原因 s);}}});rabbitTemplate.convertAndSend(MQConfig.EXCHANGE_NAME,test.hello,测试springboot整合交换机);Thread.sleep(200);}
}结果
1.2 return回退模式的代码编写
第一步开启回退模式
spring:rabbitmq:password: heimausername: heimaport: 5673virtual-host: itcasthost: 1.12.244.105#开启确认模式publisher-confirm-type: correlated#开启回退模式publisher-returns: true第二步编写returnCallBack()函数 第三步设置exchange处理消息的模式 ①setMandatory为true如果消息没有到队列queue则返回消息给发送方 ②setMandatory为false如果消息没有到队列queue则丢弃消息默认
package com.rabbitmq.springboot_mqproducer;import com.rabbitmq.springboot_mqproducer.rabbitMQconfig.MQConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.propertyeditors.StringArrayPropertyEditor;
import org.springframework.boot.test.context.SpringBootTest;import javax.annotation.Resource;SpringBootTest
class SpringbootMqProducerApplicationTests {ResourceRabbitTemplate rabbitTemplate;Testvoid contextLoads() throws InterruptedException {//编写confirm回调函数rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/**** param correlationData 相关的配置信息* param b 消息是否发送成功* param s 消息发送失败原因*/Overridepublic void confirm(CorrelationData correlationData, boolean b, String s) {System.out.println(confirm方法被执行了);System.out.println(b);if(b){System.out.println(消息从producer - exchange成功);System.out.println(失败原因 s);}else{//消息发送失败需要做一些处理System.out.println(消息从producer - exchange失败);System.out.println(失败原因 s);}}});//编写return回调函数rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {System.out.println(return回退模式回调函数执行了);System.out.println(消息returnedMessage.getMessage());System.out.println(exchangereturnedMessage.getExchange());System.out.println(replyCodereturnedMessage.getReplyCode());System.out.println(replyTextreturnedMessage.getReplyText());System.out.println(routingKeyreturnedMessage.getRoutingKey());}});//设置回退模式中exchange处理消息的方式/*当将mandatory设置为false默认值如果RabbitMQ无法将消息路由消息将会被静默丢弃生产者不会收到通知。当设置mandatory为true时意味着消息被视为mandatory如果在发布消息时RabbitMQ无法将消息路由到任何队列例如由于没有匹配的队列与指定的路由键则代理将通过调用ReturnListener回调的returnedMessage方法将消息返回给生产者发布者。生产者可以根据需要适当地处理这个返回的消息例如记录日志或执行某些恢复操作。*/rabbitTemplate.setMandatory(true);//TODO 这里把routingKey写错是为了让交换机找不到queue从而触发returnCallBack()函数rabbitTemplate.convertAndSend(MQConfig.EXCHANGE_NAME,testtttt.hello,测试springboot整合交换机);Thread.sleep(200);}}消息的可靠投递小结
设置配置publisher-confirm-type: correlated开启确认模式使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack如果为true, 则发送成功如果为false,则发送失败需要处理。设置ConnectionFactory的publisher-returnstrue开肩退回模式。使用rabbitTemplate.setReturnCallback设置退回函数当消息从exchange路由到queue失败后如果设置了rabbitTemplate.setMandatory(true)参数则会将消息退回给producer。并执行回调函数returnedMessage。在RabbitMQ中也提供了事务机制但是性能较差此处不做讲解。 使用channel下列方法完成事务控制: txSelect(),用于将当前channel设置成transaction模式 txCommit()用于提交事务 txRollback(),用于回滚事务
2.Consumer Ack
ack指Acknowledge,确认。表示消费端收到消息后的确认方式。 有三种确认方式
自动确认acknowledge“none”手动确认acknowledge“manual”根据异常情况确认acknowledge“auto”这种方式很麻烦不做讲解
其中自动确认是指当消息一旦被Consumer接收到 则自动确认收到并将相应message从RabbitMQ的消息缓存中移除。但是在实际业务处理中很可能消息接收到业务处理出现异常,那么该消息就会丢失。如果设置了手动确认方式则需要在业务处理成功后调用channel.basicAck() 手动签收如果出现异常则调用channel.basicNack() 方法让其自动重新发送消息。
代码编写 发送消息的生产者端代码不用变只需要能够发送消息就行 消费者端 第一步编写yml配置文件
spring:rabbitmq:username: heimapassword: heimavirtual-host: itcasthost: 1.12.244.105port: 5673#设置消息为手动签收listener:simple:acknowledge-mode: manual #消费者端确认模式:none自动确认 manual手动确认 auto通过抛出异常的类型来做响应的处理concurrency: 1 #当前监听的数量max-concurrency: 5 #最大监听数量retry:enabled: true #是否支持重试max-attempts: 4 #最大重试次数默认为3第二步编写消费者代码 消费者端创建一个listener并实现ChannelAwareMessageListener接口其实也可以不实现该接口只要 RabbitListener 标记的方法或者 RabbitListener 标记的类 RabbitHandler 标记的方法的参数列表有[com.rabbitmq.client.Channel]和[org.springframework.amqp.core.Message]两个参数都可以
package com.rabbit.springboot_mqconsumer;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;/*** author Watching* * date 2023/7/19* * Describe:*/
Component
public class RabbitMQListener implements ChannelAwareMessageListener {
// RabbitListener(queues {boot_topic_queue})//填写队列名称,可以以字符串数组的方式监听多个队列
// public void listener(Message message){
// System.out.println(messagemessage);
// }/*** 使用ChannelAwareMessageListener监听器接口中的onMessage()方法来充当消费者如果上面注释的方法与当前方法同时存在一条消息只会被消费一次。不会被两个方法都消费** param message* param channel* throws Exception Consumer ACK机制:* 1.设置手动签收。acknowledge manual”* 2.让监听器类实现ChannelAwareMessageListener接口* 3.如果消息成功处理则调用channel的basicAck()签收* 4.如果消息处理失败则调用channel的basicNack( )拒绝签收broker重新发送给consumer*/RabbitListener(queues boot_topic_queue )Overridepublic void onMessage(Message message, Channel channel) throws Exception {try{//1.接收消息System.out.println(message message);System.out.println(channel channel);//2.处理业务逻辑System.out.println(模拟处理业务逻辑......);//3.手动签收/*void basicAck(long deliveryTag, boolean multiple) throws IOException;deliveryTag:当消费者接收到一条消息后RabbitMQ 会为该消息分配一个唯一的 DeliveryTag。这个 DeliveryTag 是一个64位的长整型数值并且只在该 Channel 内唯一即相同 Channel 下的 DeliveryTag 不会重复。multiple:当 multiple 设置为 false 时表示只确认当前指定的 deliveryTag 对应的一条消息。也就是说只确认指定的单个消息已经成功被处理或处理失败。当 multiple 设置为 true 时表示确认当前指定的 deliveryTag 及其之前所有未确认的消息在同一个 Channel 下。也就是说会一次性确认多条消息的处理状态将 deliveryTag 小于或等于指定 deliveryTag 的所有消息都确认处理了。这种批量确认的机制有助于提高消息的处理效率特别是当消费者处理多条消息时可以通过一次性确认多条消息的方式来减少网络开销和消费者端的负担。在使用 channel.basicAck(deliveryTag, multiple) 和 channel.basicNack(deliveryTag, multiple, requeue) 方法时可以根据实际场景来选择是单条确认还是批量确认以满足不同的业务需求。*/channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);System.out.println(完成手动签收);}catch(Exception e){//4.出现异常拒绝签收/*deliveryTag一个唯一标识消息的64位长整型数值用于确认消息的消费状态。multiple一个布尔类型的参数用于决定是否批量处理多条消息。若设置为 true则会否定当前指定 deliveryTag 及其之前的所有未确认消息若设置为 false则只否定当前指定 deliveryTag 对应的一条消息。requeue一个布尔类型的参数表示是否将消息重新放回队列。若设置为 true则消息会被重新入队列RabbitMQ 会再次将它发送给消费者若设置为 false则消息会被直接丢弃不会重新放回队列。*/System.out.println(代码逻辑出现异常拒收);channel.basicNack(message.getMessageProperties().getDeliveryTag(),true,true);}}
}只需要两步就可以实现Consumer ack下面我们来测试一下 首先是正常运行的代码的结果业务逻辑代码无异常 生产者端是用的前面测试boot整合的代码 然后我们来测试业务逻辑代码出错的情况我们在业务逻辑代码处添加一个除数不能为0的异常 再次运行代码一直在重试一直再报错
消息的可靠性总结
1.持久化
exchange要持久化queue要持久化message要持久化
2.生产方确认Confirm在后续文章中会讲解如何在回调函数中进行具体的处理 3.消费方确认Ack 4. Broker高可用集群搭建
3.消费端限流
在A系统中每秒最多只能处理1000条请求如果在一秒钟只能瞬间有5000条请求打入A系统那么A系统就会崩溃所以我们在A系统中加入一个MQ中间件让5000个请求先发送到MQ然后A系统再分批次的从MQ中拉取1000条请求这样A系统就避免了崩溃的情况。 这也是我们常说的MQ的削峰功能 设置MQ消费限流很简单只需要设置两个属性
确认模式设置为手动确认在上面的Ack我们已经讲过设置prefetch属性prefetch nn就是每次从MQ中获取消息的数量 其余的消费端代码和生产者端代码不用修改。 当设置了消费端限流后如果从MQ中取出1条消息消费者端没有进行确认那么消费者端将不会再从MQ中取消息直到消息被确认。
4.TTL
TTL全称Time To Live(存活时间/过期时间)。 当消息到达存活时间后还没有被消费会被自动清除。 RabbitMQ可以对消息设置过期时间也可以对整个队列Queue设置过期时间。 举一个例子 生活中我们在购买商品的时候会下订单系统会提示我们要在30分钟之内付款否则订单将会被取消。
Ⅰ、先在控制台模拟上面的情况
①创建一个交换机 ②创建一个队列 ③进入交换机exchange_ttl和队列queue_ttl进行绑定 ④消息的发布 ⑤在消息队列中查看 将鼠标放上ttl就可以看到设置的时间等时间一过这条消息就会被自动清除。
Ⅱ、代码实现队列过期和消息过期
①创建交换机队列以及绑定关系
package com.rabbitmq.springboot_mqproducer.rabbitMQconfig;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;/*** author Watching* * date 2023/7/18* * Describe:*/
Configuration
public class MQConfig {public static final String QUEUE_TTL_NAME queue_ttl;public static final String EXCHANGE_TTL_NAME exchange_ttl;/*
创建队列测试ttl特性*/Bean(test_queue_ttl)public Queue ttlQueue() {MapString,Object arguments new HashMap();arguments.put(x-message-ttl,10000);//消息过期的时间arguments.put(x-expires,100000);//队列过期的时间//设置队列的ttl时间return QueueBuilder.durable(QUEUE_TTL_NAME).withArguments(arguments).build();//参数的属性可以在控制台上查看}/*
创建一个交换机测试队列ttl特性*/Bean(test_exchange_ttl)public Exchange ttlExchange() {return ExchangeBuilder.topicExchange(EXCHANGE_TTL_NAME).durable(true).build();}/*绑定ttl交换机和队列*/Beanpublic Binding ttlBinding(Qualifier(test_exchange_ttl) Exchange exchange, Qualifier(test_queue_ttl) Queue queue) {return BindingBuilder.bind(queue).to(exchange).with(ttl.#).noargs();}
}在创建队列时我们指定了x-message-ttl使队列中的所有消息都是一个固定的时间过期 我们还可以在发送消息时指定每条消息的过期时间。 只需要在发送方法convertAndSend()方法中添加一个消息后处理参数即可
/*MessagePostProcessor 是 Spring AMQP 中的一个接口用于对消息进行后处理。通过实现该接口你可以在发送消息之前对消息进行一些自定义处理例如添加自定义的消息头、修改消息内容等。*/MessagePostProcessor messagePostProcessor new MessagePostProcessor() {Overridepublic Message postProcessMessage(Message message) throws AmqpException {//1.设置消息属性message.getMessageProperties().setExpiration(5000);//5000ms过期//2.返回该消息return message;}};Testvoid testSend() throws InterruptedException {for (int i 0; i 10; i) {rabbitTemplate.convertAndSend(MQConfig.EXCHANGE_TTL_NAME, ttl.hello, 测试ttli,messagePostProcessor);}Thread.sleep(200);}小细节 ①当队列设置了x-expires和x-messgae-ttl消息过期时间以短的为准 ②当队列设置了x-messgae-ttl且发送消息时通过消息后处理也设置了过期时间那么消息过期时间也以短的为准。 ③当十条消息中只有一条消息设置了过期时间这条消息过期后只有处于队列顶端即即将被消费时才会对这条消息是否过期做判断。
5.死信队列
5.1 概念
死信队列英文缩写: DLX Dead Letter Exchange (死信交换机)当消息成为Dead message后,可以被重新发送到另一个交换机这个交换机就是DLX。死信队列为什么英文翻译过来使死信交换机呢因为交换机概念只有在RabbitMQ中才有其它MQ中间件只有队列概念所以习惯叫死信队列而RabbitMQ中存在交换机概念所以叫死信交换机。 在这里我们需要理解的问题有 ①消息什么时候成为死信
队列长度达到限制比如队列最多容纳10条消息当第11条消息进入时这条消息就成为了死信消息。消费者拒接消费消息basicNack/basicReject,并且不把消息重新放入原目标队列,requeuefalse;原队列存在消息过期设置消息到达超时时间却并未被消费
以上三种满足一条即为死信消息
②队列如何绑定死信交换机 队列设置参数x-dead-letter-exchange和x-dead-letter-routing-key x-dead-letter-exchange死信交换机的名称 x-dead-letter-routing-key消息发送时指定的routingKey 5.2 代码实现死信队列
创建死信队列:
1.声明正常的队列(test_queue_dLx)和交换机(test_exchange_dlx)2.声明死信队列(queue_dlx)和死信交换机(exchange_dlx)3.正常队列绑定死信交换机,正常队列绑定死信队列不需要创建Binding Bean,只需要在正常队列创建时设置参数就可以 – 设置两个参数: x-dead-letter-exchange:死信交换机名称 x-dead-letter-routing-key:发送给死信交换机的routingkey
设置正常队列中的消息的过期时间x-message-ttl 设置正常队列的长度限制x-max-length
package com.rabbitmq.springboot_mqproducer.rabbitMQconfig;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;/*** author Watching* * date 2023/7/18* * Describe:*/
Configuration
public class MQConfig {/*** 测试死信队列*//*创建普通交换机和普通队列*/Bean(test_exchange_dlx)public Exchange testDlxExchange() {return ExchangeBuilder.topicExchange(test_exchange_dlx).durable(true).build();}Bean(test_queue_dlx)public Queue testDlxQueue() {MapString,Object map new HashMap();//x-dead-letter-exchange:死信交换机名称map.put(x-dead-letter-exchange,exchange_dlx);//x-dead-letter-routing-key:发送给死信交换机的routingkeymap.put(x-dead-letter-routing-key,dlx.hehe);//这个routingkey只需要满足死信交换机的路由规则就可以//设置正常队列中的消息的过期时间ttlmap.put(x-message-ttl,10000);//设置正常队列的长度限制max-lengthmap.put(x-max_length,10);return QueueBuilder.durable(test_queue_dlx).withArguments(map).build();}Beanpublic Binding binding1(Qualifier(test_exchange_dlx) Exchange exchange,Qualifier(test_queue_dlx)Queue queue){return BindingBuilder.bind(queue).to(exchange).with(test.dlx.#).noargs();}/*创建死信交换机和死信队列*/Bean(exchange_dlx)public Exchange dlxExchange() {return ExchangeBuilder.topicExchange(exchange_dlx).durable(true).build();}Bean(queue_dlx)public Queue dlxQueue() {return QueueBuilder.durable(queue_dlx).build();}Beanpublic Binding binding2(Qualifier(exchange_dlx) Exchange exchange,Qualifier(queue_dlx)Queue queue){return BindingBuilder.bind(queue).to(exchange).with(dlx.#).noargs();}/*绑定普通队列和死信交换机并不需要写一个Binding只需要在普通队列中添加参数就行*/
}
发送消息测试死信消息 1.过期时间 2.长度限制 3.消息拒收 Testvoid testDlx() {//1.过期时间
// rabbitTemplate.convertAndSend(test_exchange_dlx,test.dlx.hello,测试消息超出过期时间变成死信);//2.超出队列消息数量限制
// for (int i 0; i 20; i) {
// rabbitTemplate.convertAndSend(test_exchange_dlx, test.dlx.hello, 测试消息超出队列数量限制变成死信);
// }//3.消费端拒收rabbitTemplate.convertAndSend(test_exchange_dlx,test.dlx.hello,测试消息被拒收变成死信);}死信队列小结 1.死信交换机,死信队列和普通交换机,普通队列没有区别. 2.当消息成为死信后,如果该队列绑定了死信交换机,则消息会被重新路由到死信队列中 3.消息成为死信的三种情况
消息在队列中到达超时时间并未被消费消息在消费者端被拒收,且设置了不重回队列队列长度存在限制,消息数量超出了限制
6.延迟队列
延迟队列即消息进入队列后不会立即被消费只有到达指定时间后才会被消费。 需求: 1.下单后30分钟未支付,取消订单回滚库存。 2.新用户注册成功7天后发送短信问候。 实现方式: 1.定时器 2.延迟队列 订单系统将订单放入延迟队列种,30分钟后取出,去库存系统中判断订单是否已经支付,再进行后续的支付或者未支付操作 但是! RabbitMQ官方没有提供延迟队列,所以我们需要使用ttl死信队列构成延迟队列 普通队列设置为30min中过期,过期后消息路由到死信队列,库存系统从死信队列中取消息,这样就形成了一个延迟队列
代码实现延迟队列
1.定义正常交换机(order_exchange)和队列(order_queue)同时绑定 2.定义死信交换机(order_exchange_dlx) 和队列(order_queue_dlx)同时绑定 3.绑定正常队列和死信交换机设置正常队列过期时间为10秒 /*** 测试延迟队列*//*1.定义正常交换机(order_exchange)和队列(order_queue)*/Bean(orderQueue)public Queue orderQueue(){//3.正常队列绑定死信交换机MapString,Object map new HashMap();map.put(x-dead-letter-exchange,order_exchange_dlx);map.put(x-dead-letter-routing-key,dlx.order.hehe);//设置正常队列的消息过期时间map.put(x-message-ttl,10000);return QueueBuilder.durable(order_queue).withArguments(map).build();}Bean(orderExchange)public Exchange orderExchange(){return ExchangeBuilder.topicExchange(order_exchange).build();}Beanpublic Binding orderBinding(Qualifier(orderQueue)Queue queue,Qualifier(orderExchange)Exchange exchange){return BindingBuilder.bind(queue).to(exchange).with(order.#).noargs();}/*2.定义死信交换机(order_exchange_dlx) 和队列(order_queue_dlx)*/Bean(orderQueueDlx)public Queue orderQueueDlx(){return QueueBuilder.durable(order_queue_dlx).build();}Bean(orderExchangeDlx)public Exchange orderExchangeDlx(){return ExchangeBuilder.topicExchange(order_exchange_dlx).build();}Beanpublic Binding orderBindingDlx(Qualifier(orderQueueDlx)Queue queue,Qualifier(orderExchangeDlx)Exchange exchange){return BindingBuilder.bind(queue).to(exchange).with(dlx.order.#).noargs();}4.创建生产者发送消息 /*** 测试延迟队列*/Testvoid testDelay() throws InterruptedException {rabbitTemplate.convertAndSend(order_exchange,order.test,测试延迟队列);for (int i 10;i 0;i--){System.out.println(i...);Thread.sleep(1000);}}5.创建消费者
package com.rabbit.springboot_mqconsumer;import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;/*** author Watching* * date 2023/8/2* * Describe:*/
Component
public class OrderListener implements ChannelAwareMessageListener {RabbitListener(queues order_queue_dlx)//监听死信队列Overridepublic void onMessage(Message message, Channel channel) throws Exception {try{//1.接收messageSystem.out.println(message:message);//2.处理业务逻辑System.out.println(处理业务逻辑);System.out.println(根据订单id在数据库中查询订单状态);System.out.println(判断订单是否支付成功);System.out.println(未支付回滚库存取消订单);//3.手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}catch (Exception e){//4.业务出错拒绝签收channel.basicNack(message.getMessageProperties().getDeliveryTag(),true,true);//业务出错拒签后要将这条消息重新放回死信队列}}
}延迟队列小结 1.延迟队列指消息进入队列后可以被延迟一定时间再进行消费。 2. RabbitMQ没有提供延迟队列功能但是可以使用: TTL DLX来实现延迟队列效果。
应用问题
1.消息补偿
消息补偿机制
2.幂等性保障
幂等性保障
