行业知识
RabbitMQ队列
RabbitMQ队列
1、***信的概念
先从���念解释上搞清楚这个定义，***信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说,producer将消息投递到broker或者直接到queue里了，consumer 从 queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了***信，有***信自然就有了***信队列。
应用场景:为了保证订单业务的消息数据不丢失，需要使用到RabbitMQ的***信队列机制，当消息消费发生异常时，将消息投入***信队列中.还有比如说:用户在商城下单成功并点击去支付后在指定时间未支付时自动失效
1.1、***信的来源
- 消息TTL过期；
- 队列达到最大长度(队列满了，无法再添加数据到mq.中)；
- 消息被拒绝(basic.reject 或 basic.nack)并且requeue=false；
1.2、***信实战
架构图
1.3、消费者01
//***信队列public class DeadMessageConsumer01 { //普通交换机 public static final String NORMAL_EXCHANGE = "normal_exchange"; //***信交换机 public static final String DEAD_EXCHANGE = "dead_exchange"; //普通队列 public static final String NORMAL_QUEUE = "normal_queue"; //***信队列 public static final String DEAD_QUEUE = "dead_queue"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMQUtils.getChannel(); //声明普通和***信交换机 channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); //声明普通队列，这一次需要使用参数了 Map arguments = new HashMap(); //这里就要设置***信到***信交换机上了 arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE); //设置***信routingKey arguments.put("x-dead-letter-routing-key","deadKey"); channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments); //声明***信队列 channel.queueDeclare(DEAD_QUEUE,false,false,false,null); //开始分别将***信和普通交换机与队列绑定 channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normalKey"); channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"deadKey"); DeliverCallback deliverCallback = (consumerTag,message) ->{ System.out.println("***信测试普通队列消费者01收到消息："+new String(message.getBody())); }; channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,consumerTag -> {}); }}1.4、生产者
//***信队列的生产者public class DeadMessageProducers { //普通交换机 public static final String NORMAL_EXCHANGE = "normal_exchange"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMQUtils.getChannel(); //***信消息设置TTL过期时间，设置10秒 AMQP.BasicProperties basicProperties = new AMQP.BasicProperties() .builder().expiration("10000").build(); //发10条消息 for (int i = 1; i 先把消费者启动，将交换机和队列先声明出来，然后关掉消费者再启动生产者，生产者一直发这10个消息，但是消费者已经停止了，没人消费，所以TTL一到期就会被放进***信队列里，我们在web管理界面就能看到，普通队列的10条消息很快就跑到***信队列中了。
然后现在***信队列里有那10条消息，再来一个专门消费***信消息的消费者2，非常简单，只需要消费***信队列即可。
//***信队列public class DeadMessageConsumer02 { //***信队列 public static final String DEAD_QUEUE = "dead_queue"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMQUtils.getChannel(); DeliverCallback deliverCallback = (consumerTag,message) ->{ System.out.println("***信测试普通队列消费者02收到消息："+new String(message.getBody())); }; channel.basicConsume(DEAD_QUEUE,true,deliverCallback,consumerTag -> {}); }}启动它消费***信队列
消费完成，看看管理界面，***信消息归0
1.5、队列达到最大长度的***信
在原来的消费者基础上增加一条
//设置队列最大长度arguments.put("x-max-length",6); 原来的生产者也不需要给消息设置过期时间了直接发送消息即可。
测试之前先将原来的普通队列删除再启动消费者
仍然把消费者停掉，生产者启动，发送10条消息6条正常4条超过了限制会被放到***信队列
1.6、消息被拒***信队列
我们先把原来的消息全部消费掉或者直接删除原来的队列，再演示新的。
消费者做一些修改
DeliverCallback deliverCallback = (consumerTag,message) ->{ String msg = new String(message.getBody()); if (msg.equals("dead5")){ System.out.println("拒绝第5个消息："+ msg); //basicReject第二个参数是是否还放回队列 channel.basicReject(message.getEnvelope().getDeliveryTag(),false); } else { System.out.println("***信测试普通队列消费者01收到消息：" + msg); channel.basicAck(message.getEnvelope().getDeliveryTag(),false); }};channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,consumerTag -> {});生产者发送10条消息，看到消费者01拒绝了第5个消息，其他的正常收到
通过管理界面看到第5条消息放到了***信队列中
启动专门消费***信队列的消费者02将第5个消息消费。
2、延迟队列
2.1、延迟队列概念
延时队列,队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。
其实，TTL过期导致的***信队列就是延迟队列
应用场景：
- 1.订单在十分钟之内未支付则自动取消
- ⒉新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
- 3.用户注册成功后，如果三天内没有登陆则进行短信提醒。
- 4.用户发起退款，如果三天内没有得到处理则通知相关运营人员。
- 5.预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议
这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务，如:发生订单生成事件，在十分钟之后检查该订单支付状态，然后将未支付的订单进行关闭;看起来似乎使用定时任务，一直轮询数据，每秒查一次，取出需要被处理的数据，然后处理不就完事了吗?
如果数据量比较少，确实可以这样做，比如：对于“如果账单一周内未支付则进行自动结算”这样的需求,如果对于时间不是严格限制，而是宽松意义上的一周，那么每天晚上跑个定时任务检查一下所有未支付的账单，确实也是一个可行的方案。
但对于数据量比较大，并且时效性较强的场景，如：“订单十分钟内未支付则关闭“，短期内未支付的订单数据可能会有很多，活动期间甚至会达到百万甚至千万级别，对这么庞大的数据量仍旧使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单的检查，同时会给数据库带来很大压力，无法满足业务要求而且性能低下。
2.1.1、RabbitMQ 中的 TTL
TTL 是什么呢？TTL 是 RabbitMQ 中一个消息或者队列的属性，表明一条消息或者该队列中的所有 消息的最大存活时间， 单位是毫秒。
换句话说，如果一条消息设置了 TTL 属性或者进入了设置 TTL 属性的队列，那么这 条消息如果在 TTL 设置的时间内没有被消费，则会成为"***信"。如果同时配置了队列的 TTL 和消息的 TTL，那么较小的那个值将会被使用，有两种方式设置 TTL。
-
消息设置 TTL
例如：
rabbitTemplate.convertAndSend("X","XC",message,msg -> { //发送消息设置消息的TTL msg.getMessageProperties().setExpiration(ttl+"000"); return msg; }); -
**队列设置 TTL **
例如：在创建队列的时候设置队列的“x-message-ttl”属性
//声明普通队列 TTL 为40秒 @Bean("queueB") public Queue queueB(){ HashMap map = new HashMap(3); //设置***信交换机 map.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE); //设置***信RoutingKey map.put("x-dead-letter-routing-key","YD"); //设置过期时间,10秒 map.put("x-message-ttl",40000); return QueueBuilder.durable(QUEUE_B).withArguments(map).build(); }2.2、整合SpringBoot
添加依赖项
8 8 2.3.4.RELEASE 1.18.20 2.1.4.RELEASE 2.9.2 org.springframework.boot spring-boot-starter-amqp ${spring.boot.version} org.springframework.boot spring-boot-starter-web ${spring.boot.version} org.springframework.boot spring-boot-starter-test ${spring.boot.version} test com.taobao.arthas fastjson 1.2.80-fix org.projectlombok lombok ${lombok.version} io.springfox springfox-swagger2 ${springfox.version} io.springfox springfox-swagger-ui ${springfox.version} org.springframework.amqp spring-rabbit-test ${spring.amqp.version} test org.springframework.boot spring-boot-maven-plugin创建application.properties文件
spring.rabbitmq.host=192.168.111.28spring.rabbitmq.port=5672spring.rabbitmq.username=adminspring.redis.password=123
swagger2的配置类SwaggerConfig
@Configuration@EnableSwagger2public class SwaggerConfig { @Bean public Docket webApiConfig(){ return new Docket(DocumentationType.SWAGGER_2) .groupName("webApi") .apiInfo(webApiInfo()) .select() .build(); } private ApiInfo webApiInfo(){ return new ApiInfoBuilder() .title("rabbitmq 接口文档") .description("本文档描述了 rabbitmq 微服务接口定义") .version("1.0") .contact(new Contact("zm", "http://zmblog.vip", "3339332352@qq.com")) .build(); }}2.3、队列TTL
创建两个队列 QA 和 QB，两者队列 TTL 分别设置为 10S 和 40S，然后在创建一个交换机 X 和***信交 换机 Y，它们的类型都是 direct，创建一个***信队列 QD，它们的绑定关系如下：
2.4、配置文件类代码
TtlQueueConfig
//TTL队列，配置类代码@Configurationpublic class TtlQueueConfig { //普通交换机名称 public static final String X_EXCHANGE = "X"; //***信交换机名称 public static final String Y_DEAD_LETTER_EXCHANGE = "Y"; //普通队列名称，两个 public static final String QUEUE_A = "QA"; public static final String QUEUE_B = "QB"; //***信队列名称 public static final String DEAD_LETTER_QUEUE = "QD"; //声明普通交换机X_EXCHANGE，起一个别明xExchange注入bean @Bean("xExchange") public DirectExchange xExchange(){ return new DirectExchange(X_EXCHANGE); } //声明***信交换机Y_EXCHANGE，起一个别明yExchange注入bean @Bean("yExchange") public DirectExchange yExchange(){ return new DirectExchange(Y_DEAD_LETTER_EXCHANGE); } //声明普通队列 TTL 为10秒 @Bean("queueA") public Queue queueA(){ HashMap map = new HashMap(3); //设置***信交换机 map.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE); //设置***信RoutingKey map.put("x-dead-letter-routing-key","YD"); //设置过期时间,10秒 map.put("x-message-ttl",10000); return QueueBuilder.durable(QUEUE_A).withArguments(map).build(); } //声明普通队列 TTL 为40秒 @Bean("queueB") public Queue queueB(){ HashMap map = new HashMap(3); //设置***信交换机 map.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE); //设置***信RoutingKey map.put("x-dead-letter-routing-key","YD"); //设置过期时间,10秒 map.put("x-message-ttl",40000); return QueueBuilder.durable(QUEUE_B).withArguments(map).build(); } //***信队列 @Bean("queueD") public Queue queueD(){ return QueueBuilder.durable(DEAD_LETTER_QUEUE).build(); } //绑定 @Bean public Binding QABindingX(@Qualifier("queueA") Queue queueA, @Qualifier("xExchange") DirectExchange xExchange){ return BindingBuilder.bind(queueA).to(xExchange).with("XA"); } @Bean public Binding QBBindingX(@Qualifier("queueB") Queue queueB, @Qualifier("xExchange") DirectExchange xExchange){ return BindingBuilder.bind(queueB).to(xExchange).with("XB"); } @Bean public Binding QDBindingY(@Qualifier("queueD") Queue queueD, @Qualifier("yExchange") DirectExchange yExchange){ return BindingBuilder.bind(queueD).to(yExchange).with("YD"); }}生产者controller，DeadLetterProducers
//生产者发送延迟消息@Slf4j@RestControllerpublic class DeadLetterProducers { //自动注入 @Autowired private RabbitTemplate rabbitTemplate; //发消息 @GetMapping("/sendMsg/{message}") public void sendMsg(@PathVariable String message){ log.info("当前时间：{}，发送一条消息给两个TTL队列：{}",new Date().toString(),message); rabbitTemplate.convertAndSend("X","XA","消息来自于TTL为10秒的队列："+message); rabbitTemplate.convertAndSend("X","XB","消息来自于TTL为40秒的队列："+message); }}消费者：
//消费者@Slf4j@Componentpublic class DeadLetterQueueConsumer { //接收消息,定义一个监听器来监听***信队列中的消息 @RabbitListener(queues = "QD") public void receiveD(Message message, Channel channel){ String s = new String(message.getBody()); log.info("当前时间：{},收到***信队列的消息：{}",new Date().toString(),s); }}成功启动
在浏览器上输入localhost:8080/sendMsg/哈哈哈哈哈
发现后台开始输出信息，先是发送的提示信息，然后过了10秒消费者收到了一个信息，然后再过30秒收到了第二条信息。
结果是没有问题的，第一条消息在10S后变成了***信消息，然后被消费者消费掉，第二条消息在40S之后变成了***信消息，然后被消费掉，这样一个延时队列就打造完成了。
不过，如果这样使用的话，岂不是每增加一个新的时间需求，就要新增一个队列，这里只有10S和40S两个时间选项，如果需要一个小时后处理，那么就需要增加TTL为一个小时的队列，如果是预定会议室然后提前通知这样的场景，岂不是要增加无数个队列才能满足需求?
2.5、延迟队列优化
在这里新增了一个队列 QC，它是通用的延迟队列，生产者掌握时间（这是设定消息的过期时间，之前是设定队列的过期时间）,绑定关系如下,该队列不设置 TTL 时间
我们让生产者来决定延迟队列的时间，需要多少就指定多少这样更符合需求。
在TtlQueueConfig类中添加内容
public static final String QUEUE_C = "QC";//新加一个通用TTL的普通队列 @Bean("queueC") public Queue queueC(){ HashMap map = new HashMap(2); //设置***信交换机 map.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE); //设置***信RoutingKey map.put("x-dead-letter-routing-key","YD"); //TTL就交给生产者指定了这里就不需要写了 return QueueBuilder.durable(QUEUE_C).withArguments(map).build(); }//QC绑定普通交换机 @Bean public Binding QCBindingX(@Qualifier("queueC") Queue queueC, @Qualifier("xExchange") DirectExchange xExchange){ return BindingBuilder.bind(queueC).to(xExchange).with("XC"); }生产者的controller添加一个请求
//发指定TTL的消息@GetMapping("/sendExpMsg/{message}/{ttl}")public void sendExpMsg(@PathVariable String message,@PathVariable String ttl){ log.info("当前时间：{}，发送一条时长为：{}毫秒的队列消息给队列QC：{}",new Date(),ttl,message); rabbitTemplate.convertAndSend("X","XC",message,msg -> { //发送消息设置消息的TTL msg.getMessageProperties().setExpiration(ttl+"000"); return msg; });} 我们启动服务，浏览器上分别输入localhost:8080/sendExpMsg/这是第一条信息/20，localhost:8080/sendExpMsg/这是第二条信息/2
我们发现，结果并不是我们所设想的那样，延时20秒的和2秒的两条消息都是在20秒时接收到，同时接收了，而这个顺序也是按照先进先出的队列特性来的，显然这不是实际需求。
在最开始的时候，我们就了解到如果在消息属性上设置TTL的方式，消息可能并不会按时“***亡”，因为RabbitMQ只会检查第一个消息是否过期，如果过期则丢到***新信队列，如果第一个消息的延时时间非常长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行。
那么如何解决这个问题，就需要使用RabbitMQ的插件实现延迟队列
2.6、RabbitMQ的插件实现延迟队列
如果不能实现在消息粒度上的TTL，并使其在设置的TL时间及时***亡，就无法设计成一个通用的延时队列。
2.6.1、安装RabbitMQ的延时队列插件
在官网上下载https://www.rabbitmq.com/community-plugins.html，rabbitmq_delayed_message__exchange插件，然后解压放置到RabbitMQ的插件目录。
下载3.8.0的连接rabbitmq_delayed_message_exchange-3.8.0.ez (github.com)
进入 RabbitMQ 的安装目录下的 plgins 目录，执行下面命令让该插件生效，然后重启 RabbitMQ
/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/pluginsrabbitmq-plugins enable rabbitmq_delayed_message_exchange
把插件放到RabbitMQ的插件文件夹
然后进行安装
重启Rabbitmq之后就可以了，可以看到交换机的类型新加一个x-delayed-message类型的
这样延迟的任务就交给交换机来做了，生产者把消息发给交换机，然后消息就在交换机延迟了，然后延迟时间过了之后就可以发给普通的队列了。
2.6.2、基于插件的延迟消息实现
我们新加一个队列delayed.queue,一个自定义交换机 delayed.exchange，绑定关系如下:
在我们自定义的交换机中，这是一种新的交换类型，该类型消息支持延迟投递机制 消息传递后并 不会立即投递到目标队列中，而是存储在 mnesia(一个分布式数据系统)表中，当达到投递时间时，才 投递到目标队列中。
配置文件类
//使用插件实现延迟public class DelayedQueueConfig { //声明延迟类型的交换机名称 public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange"; //队列 public static final String DELAYED_QUEUE_NAME = "delayed.queue"; //routingKey public static final String DELAYED_ROUTING_KEY = "delayed.routingKey"; //声明队列 @Bean public Queue delayedQueue(){ return QueueBuilder.durable(DELAYED_QUEUE_NAME).build(); } //声明交换机CustomExchange是自定义交换机，由于使用插件提供的类型在原来的四个类型中没有所以只能自定义 /* public CustomExchange(String name, String type, boolean durable, boolean autoDelete, Map arguments) { super(name, durable, autoDelete, arguments); this.type = type; } 第一个参数就是交换机的名称 第二个参数就是交换机的自定义类型 第三个参数就是是否持久化 第四个参数就是是否自动删除 第五个是其他参数 * */ @Bean public CustomExchange delayedExchange(){ HashMap arguments = new HashMap(); //设置自定义交换机的类型 //消息确实是延迟了,但是怎么传播到队列呢,是要多播呢还是直连呢,所以还得设置类型 //这里设置直连，怎么发送是直连的发送，啥时候发送就是延迟。 arguments.put("x-delayed-type","direct"); return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message", true,false,arguments); } //绑定 //返回中最后的构建，build有参数，noargs没有参数 @Bean public Binding delayedBinding(@Qualifier("delayedQueue") Queue delayedQueue, @Qualifier("delayedExchange") CustomExchange delayedExchange){ return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs(); }} 生产者controller新加请求方法
//延迟插件发送消息@GetMapping("/sendDelayMsg/{message}/{delayTime}")public void sendDelayMsg(@PathVariable String message,@PathVariable int delayTime){ log.info("当前时间：{}，发送一条时长为：{}秒的队列消息延迟队列delay.queue：{}",new Date(),delayTime,message); rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME, DelayedQueueConfig.DELAYED_ROUTING_KEY,message, msg -> { //发送消息设置延时时长 msg.getMessageProperties().setDelay(delayTime*1000); return msg; });} 消费者
//专门消费由延迟交换机发出的消息@Slf4j@Componentpublic class DelayedLetterConsumer { @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME) public void receiveDelayed(Message message){ String s = new String(message.getBody()); log.info("当前时间：{},收到delayed.queue队列的消息：{}",new Date(),s); }}我们启动服务浏览器输入localhost:8080/sendDelayMsg/这是第一条信息/20，localhost:8080/sendDelayMsg/这是第二条信息/2
可以看到结果是很正确的，第二条消息在发送2秒后就收到了，第一条消息在它发送20秒后收到的
2.6.3、总结
延时队列在需要延时处理的场景下非常有用，使用RabbitMQ来实现延时队列可以很好的利用RabbitMQ的特性，如:消息可靠发送、消息可靠投递、***信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过RabbitMQ集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。
当然，延时队列还有很多其它选择，比如利用Java的DelayQueue，利用Redis的zset，利用Quartz或者利用kafka的时间轮，这些方式各有特点,看需要适用的场景。
-
