注意：由于队列的先进先出特性，只有过期的消息到了队列的顶端（队首），才会被真正的丢弃或者进入延迟队列。所以再开律师用RabbitMQ来实现延迟队列任务的时候，需要确保业务上每个任务的延迟时间是一致的。如果遇到不同的任务类型需要不同的延迟时间的话，需要为每一种不同延迟时间的消息建立单独的消息队列。
一：RabbitMQ整合springboot
新建项目
导入下面依赖
<dependencies>
<!--RabbitMQ 依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.47</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<!--swagger-->
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>2.9.2</version>
</dependency>
<!--RabbitMQ 测试依赖-->
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
添加 Swagger 配置类
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.service.Contact;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;
@Configuration
@EnableSwagger2
public 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("enjoy6288", "http://atguigu.com",
"1551388580@qq.com"))
.build();
} }
修改配置文件
spring.rabbitmq.host=182.92.234.71
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123
二：RabbitMQ的延迟队列
1.延迟队列的概念
延时队列,队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。
2.延迟队列的实战
mq配置类：用于声明交换机和队列
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;
/**
* TTL队列 配置文件类代码
*/
@Configuration
public class TtlQueueConfig {
// 声明普通交换机
public static final String X_EXCHANGE = "X";
// 声明死信交换机
public static final String Y_DEAF_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";
// 声明xExchange
@Bean("xExchange")
public DirectExchange xExchange(){
return new DirectExchange(X_EXCHANGE);
}
// 声明yExchange
@Bean("yExchange")
public DirectExchange yExchange(){
return new DirectExchange(Y_DEAF_LETTER_EXCHANGE);
}
// 声明队列
@Bean("queueA")
public Queue queueA(){
HashMap<String, Object> arguments = new HashMap<>(3);
// 设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAF_LETTER_EXCHANGE);
// 设置死信RoutingKey
arguments.put("x-dead-letter-routing-key","YD");
//设置过期时间
arguments.put("x-message-ttl",10000);
return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
}
// 声明队列
@Bean("queueB")
public Queue queueB(){
HashMap<String, Object> arguments = new HashMap<>(3);
// 设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAF_LETTER_EXCHANGE);
// 设置死信RoutingKey
arguments.put("x-dead-letter-routing-key","YD");
//设置过期时间
arguments.put("x-message-ttl",40000);
return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
}
// 死信队列
@Bean("queueD")
public Queue queueD(){
return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
}
// 绑定
@Bean
public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
@Qualifier("xExchange") DirectExchange xExchange){
return BindingBuilder.bind(queueA).to(xExchange).with("XA");
}
// 队列b绑定B交换机
@Bean
public Binding queueBBindX(@Qualifier("queueB")Queue queueB,
@Qualifier("xExchange") DirectExchange xExchange){
return BindingBuilder.bind(queueB).to(xExchange).with("XB");
}
@Bean
public Binding queueDBindY(@Qualifier("queueD")Queue queueD,
@Qualifier("yExchange") DirectExchange yExchange){
return BindingBuilder.bind(queueD).to(yExchange).with("YD");
}
}
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Date;
/**
* 队列TTL 消费者
*
*/
@Slf4j
@Component
public class DeadLetterQueueConsumer {
// 接收消息
@RabbitListener(queues = "QD")
public void receiveD(Message message, Channel channel) throws IOException {
String msg = new String(message.getBody());
log.info("当前时间：{}，收到死信队列的消息:{}",new Date().toString(),msg);
}
}
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;
/**
*
*/
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {
@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未10s的队列"+message);
rabbitTemplate.convertAndSend("X","XB","消息来自ttl未40s的队列"+message);
}
}
访问：http://localhost:8080/ttl/sendMsg/我爱你该网址
3.延迟队列的优化
可以发送任意时间需求的消息
新建一个队列QC，按照QA和QB的方式配置，但不再配置TTL（过期）时间
在controller层新增一个接口
// 开始发消息 消息 TTL
@GetMapping("/sendExpirationMsg/{message}/{ttlTime}")
public void sendMsg(@PathVariable String message,
@PathVariable String ttlTime){
log.info("当前按=时间：{},发送一条消息时长{}毫秒TTL信息给队列QC:{}",new Date().toString(),ttlTime,message);
rabbitTemplate.convertAndSend("X","XC",message,(msg)->{
msg.getMessageProperties().setExpiration(ttlTime);
return msg;
});
}
http://localhost:8080/ttl/sendExpirationMsg/你好1/20000
http://localhost:8080/ttl/sendExpirationMsg/你好2/2000
看起来似乎没什么问题，但是在最开始的时候，就介绍过如果使用在消息属性上设置 TTL 的方式，消息可能并不会按时“死亡“，因为 RabbitMQ 只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行。
4.基于插件的延迟队列
安装rabbitmq_delayed_message_exchange 插件，详细请看：windows下rabbitmq安装延时插件（处理各种问题）
安装插件之后交换机负责延迟，而不是队列负责延迟，队列只负责接收消息
配置延迟队列：
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
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;
@Configuration
public class DelayedQueueConfig {
//延迟队列
public static final String DELAYED_QUEUE_NAME = "delayed.queue";
//延迟交换机
public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
// 声明routingKey
public static final String DELAYED_ROUTING_KEY = "delayed.routingKey";
@Bean("delayedQueue")
public Queue delayedQueue(){
return new Queue(DELAYED_QUEUE_NAME);
}
// 声明交换机
@Bean("delayedExchange")
public CustomExchange delayedExchange(){
Map<String, Object> arguments = new HashMap<>();
// 定义交换机的类型a
arguments.put("x-delayed-type","direct");
return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",true,false,arguments);
}
@Bean
public Binding delayedQueueBindingDelayedExchange(@Qualifier("delayedQueue") Queue delayedQueue,
@Qualifier("delayedExchange") CustomExchange delayedExchange){
return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
}
}
发送消息：
// 开始发消息 基于插件的
@GetMapping("/sendDelayMsg/{message}/{delayedTime}")
public void sendDelayedMsg(@PathVariable String message,
@PathVariable Integer delayedTime){
log.info("当前时间：{}，发送一条市场{}毫秒信息给延迟队列delayed.queue:{}",
new Date().toString(),delayedTime,message);
rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME,
DelayedQueueConfig.DELAYED_ROUTING_KEY,message,msg->{
msg.getMessageProperties().setDelay(delayedTime);
return msg;
});
}
基于插件的消费者
import com.rabbitmq.springboot.config.DelayedQueueConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
* 消费者 基于插件的消费者
*/
@Slf4j
@Component
public class DelayedQueueConsumer {
@RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
public void receiveDelayedQueue(Message message){
String msg = new String(message.getBody());
log.info("当前时间：{}，收到延迟队列的消息：{}",new Date().toString(),msg);
}
}
测试方法：
http://localhost:8080/ttl/sendDelayMsg/come on baby1/20000
http://localhost:8080/ttl/sendDelayMsg/come on baby2/2000
延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。
当然，延时队列还有很多其它选择，比如利用 Java 的 DelayQueue，利用 Redis 的 zset，利用 Quartz或者利用 kafka 的时间轮，这些方式各有特点,看需要适用的场景。
三：发布确认高级
生产者发送消息给rabbitmq，如果此时mq挂掉了，消息该如何保存。
配置类：
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class confirmConfig {
public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
// 声明业务exchange
@Bean("confirmExchange")
public DirectExchange confirmExchange(){
return new DirectExchange(CONFIRM_EXCHANGE_NAME);
}
// 声明确认队列
@Bean("confirmQueue")
public Queue confirmQueue(){
return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
}
// 声明确认队列绑定关系
@Bean
public Binding queuBinding(@Qualifier("confirmExchange") DirectExchange confirmExchange,
@Qualifier("confirmQueue") Queue confirmQueue){
return BindingBuilder.bind(confirmQueue).to(confirmExchange).with("key1");
}
}
生产者和消费者
@RestController
@Slf4j
@RequestMapping("/confirm")
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;
@GetMapping("/sendMessage/{message}")
public void sendMessage(@PathVariable String message){
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_QUEUE_NAME,message);
log.info("发送消息内容：{}",message);
}
}
import com.rabbitmq.springboot.config.ConfirmConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class Consumer {
@RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
public void receiveConfirmMessage(Message message){
String msg = new String(message.getBody());
log.info("接收道德队列confirm.queue消息：{}",msg);
}
}
回调接口（交换机确认）
package com.rabbitmq.springboot.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct // 注入，在执行初始化方法之前调用，
public void init(){
rabbitTemplate.setConfirmCallback(this);
}
/**
* 交换机确认回调方法
* 发消息 交换机收到了 回调
* @param correlationData 保存回调消息的ID及相关信息
* @param ack 交换机收到消息 true
* @param cause null 交换机没有收到消息，这里方式的失败的原因
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
String id = correlationData != null ? correlationData.getId() : "";
if(ack){
log.info("交换机已经收到id为：{}的消息",id);
}else {
log.info("交换机未收到id为：{}的消息,原因是{}",id,cause);
}
}
}
注意在application.properties上加入：
spring.rabbitmq.publisher-confirm-type=correlated
在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。那么如何让无法被路由的消息帮我想办法处理一下？最起码通知我一声，我好自己处理啊。通过设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。
回退消息处理：
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback ,RabbitTemplate.ReturnCallback{
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct // 注入，在执行初始化方法之前调用，
public void init(){
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnCallback(this);
}
/**
* 交换机确认回调方法
* 发消息 交换机收到了 回调
* @param correlationData 保存回调消息的ID及相关信息
* @param ack 交换机收到消息 true
* @param cause null 交换机没有收到消息，这里方式的失败的原因
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
String id = correlationData != null ? correlationData.getId() : "";
if(ack){
log.info("交换机已经收到id为：{}的消息",id);
}else {
log.info("交换机未收到id为：{}的消息,原因是{}",id,cause);
}
}
// 可以当消息传递过程中不可达目的地是将消息返回生产者
@Override
public void returnedMessage(Message message, int replyCode, String replyText,
String exchange, String routingKey) {
log.error("消息{}，被交换机{}退回，退回原因：{}，路由keyL:{},",message,exchange,replyText,routingKey);
}
}
import com.rabbitmq.springboot.config.ConfirmConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@Slf4j
@RequestMapping("/confirm")
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;
@GetMapping("/sendMessage/{message}")
public void sendMessage(@PathVariable String message){
CorrelationData correlationData1 = new CorrelationData("1");
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,
ConfirmConfig.CONFIRM_QUEUE_NAME,message,correlationData1);
log.info("发送消息内容：{}",message+"key1");
CorrelationData correlationData2 = new CorrelationData("2");
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME+"2",
ConfirmConfig.CONFIRM_QUEUE_NAME,message,correlationData2);
log.info("发送消息内容：{}",message+"key2");
}
}
消息回退结果：访问http://localhost:8080/confirm/sendMessage/我爱你
四：备份交换机
当现有的交换机挂掉之后，所有的消息发送给备份交换机。
备份
交换机可以理解为 RabbitMQ 中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout ，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。
声明备份交换机和报警队列的关系
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ConfirmConfig {
public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
public static final String BACKUP_EXCHANGE_NAME = "backup.exchange";
public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
public static final String BACKUP_QUEUE_NAME = "backup.queue";
public static final String WARNING_QUEUE_NAME = "warning.queue";
// 声明业务exchange
@Bean("confirmExchange")
public DirectExchange confirmExchange(){
// 设置备份交换机
return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
.withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
}
// 声明确认队列
@Bean("confirmQueue")
public Queue confirmQueue(){
return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
}
// 声明确认队列绑定关系
@Bean
public Binding queuBinding(@Qualifier("confirmExchange") DirectExchange confirmExchange,
@Qualifier("confirmQueue") Queue confirmQueue){
return BindingBuilder.bind(confirmQueue).to(confirmExchange).with("key1");
}
// 备份交换机
@Bean("backupExchange")
public FanoutExchange backupExchange(){
return new FanoutExchange(BACKUP_EXCHANGE_NAME);
}
@Bean("backupQueue")
public Queue backupQueue(){
return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
}
@Bean("warningQueue")
public Queue warningQueue(){
return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
}
@Bean
public Binding backupQueueBingdingBackupExchange(@Qualifier("backupExchange") FanoutExchange backupExchange,
@Qualifier("backupQueue") Queue backupQueue){
return BindingBuilder.bind(backupQueue).to(backupExchange);
}
@Bean
public Binding warningQueueBingdingBackupExchange(@Qualifier("backupExchange") FanoutExchange backupExchange,
@Qualifier("warningQueue") Queue warningQueue){
return BindingBuilder.bind(warningQueue).to(backupExchange);
}
}
报警消费者
import com.rabbitmq.springboot.config.ConfirmConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* 报警消费者
*/
@Slf4j
@Component
public class WarningConsumer {
@RabbitListener(queues = ConfirmConfig.WARNING_QUEUE_NAME)
public void receiveWarningMsg(Message message){
String msg = new String(message.getBody());
log.info("发现报警不可用路由消息:{}",msg);
}
}
访问：http://localhost:8080/confirm/sendMessage/我爱你
结果：
当回退消息和备份交换机同时使用时，回退消息不会起作用。