之前一直没使用过RabbitMQ，最近有一个需求需要用到延迟消息，就简单的使用了一下，发现还是有蛮多坑的。
此篇文章只是RabbitMQ延迟消息相关内容，至于安装RabbitMQ等其它操作，参考百度。
顾问思议所谓延迟消息就是延迟消息！！！
在RabbitMQ中实现延迟消息有下面几种方式。
原生的RabbitMQ是不支持延迟消息的，我们可以先在MQ上安装一个插件然后再发送延迟消息。
优点：发送延迟消息比较简单，安装之后就相当于MQ支持了延迟消息
缺点：要安装插件
可以理解成，给A队列发送一个设置了过期时间的消息（过期分为消息过期和队列过期），但是不给A队列设置消费者，这个A队列就是死信队列（其实就是一个普通队列，不给它设置消费者而已）。
等到了时间还没有被消费，这个消息就会被投递到配置的队列，我们可以给这个队列设置一个消费者，然后去消费这个消息就可以达到延迟的目的。
绿色的线表示正常的消息流程，红色表示投递到死信队列流程
注：
优点： 听起来牛X，不需要安装插件
缺点：写起来复杂，还浪费一个队列
一般都是在自己的服务器上去安装，但有的公司也会去买阿里云的RabbitMQ
阿里云版的除了上述的两种还额外的支持了延迟消息，你只要设置消息头即可
官方文档 https://help.aliyun.com/document_detail/148083.html#section-rpd-mjh-kee
优点：什么都不用做，直接用
缺点：要钱
自行百度，如果你没有安装插件，使用这个延迟类型的交换机会报错
Channel shutdown: connection error; protocol method: #method<connection.close>(reply-code=503, reply-text=COMMAND_INVALID - unknown exchange type 'x-delayed-message', class-id=40, method-id=10)
// 配置延时交换器
@Bean
public CustomExchange delayExchange() {
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
return new CustomExchange("delay_exchange_name", "x-delayed-message",true, false,args);
}
消息都是一样的，在下面演示
插件和阿里云的实现方式都很简单，一个是在配置交换机的时候设置一个属性，一个是直接发消息的时候设置过期时间，就不把全部的代码写出来了。
但是死信队列相对来说复杂一些，这里就把全部的代码罗列出来帮助你理解。
队列、交换机、队列和交换机绑定配置
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class CityMapQueueConfig {
// 定义普通队列和死信队列的名字
public static final String CITY_MAP_QUEUE = "city_map_queue";
public static final String CITY_MAP_DEAD_QUEUE = "city_map_dead_queue";
// 定义普通交换机和死信交换机的名字
public static final String CITY_MAP_DEAD_NAME = "city_map_dead_exchange";
public static final String CITY_MAP_NAME = "city_map_exchange";
// 定义普通路由key和死信路由key的名字
public static final String CITY_MAP_ROUTE_KEY = "city_map_route_key";
public static final String CITY_MAP_ROUTE_DEAD_KEY = "city_map_route_dead_key";
// 创建普通队列
@Bean
public Queue cityMapQueue() {
return QueueBuilder
.durable(CITY_MAP_QUEUE)
.build();
}
// 创建死信队列
@Bean
public Queue cityMapDeadQueue() {
return QueueBuilder
.durable(CITY_MAP_DEAD_QUEUE)
// 给死信队列配置 死信交换机 和 死信路由key
.withArgument("x-dead-letter-exchange", CITY_MAP_NAME)
.withArgument("x-dead-letter-routing-key", CITY_MAP_ROUTE_KEY)
// 设置队列过期时间
// .withArgument("x-message-ttl", 1000 * 15)
.build();
}
// 创建普通交换机
@Bean
public Exchange exchange() {
return ExchangeBuilder
.topicExchange(CITY_MAP_NAME)
.durable(true)
.build();
}
// 创建死信交换机
@Bean
public Exchange deadExchange() {
return ExchangeBuilder
.topicExchange(CITY_MAP_DEAD_NAME)
.durable(true)
.build();
}
// 绑定死信交换机和队列
@Bean
public Binding deadBinding() {
return BindingBuilder
.bind(cityMapDeadQueue())
.to(deadExchange())
.with(CITY_MAP_ROUTE_DEAD_KEY)
.and(null);
}
// 绑定普通交换机和队列
@Bean
public Binding binding() {
return BindingBuilder
.bind(cityMapQueue())
.to(exchange())
.with(CITY_MAP_ROUTE_KEY)
.and(null);
}
}
消费者，只是需要消费一个队列就好了，死信队列无需设置消费者
import cn.ideamake.yxproperty.docking.mq.config.CityMapQueueConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 城市地图消费者
*
* @author 陶梓洋
* @date 2022/03/22
*/
@Component
@Slf4j
public class CityMapConsumer {
@RabbitListener(queues = CityMapQueueConfig.CITY_MAP_QUEUE)
@RabbitHandler
public void handler(Message message) {
log.info("接受到MQ异步消息");
// TODO
}
}
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
@Component
@Slf4j
public class RabbitMqProvider {
@Resource
private AmqpTemplate rabbitMqTemplate;
/**
* 发送延迟消息
* @param exchange 死信交换机
* @param routerKey 死信路由key
* @param jsonParams json格式的消息
* @param messageId 唯一消息id
* @param delay 延迟时间 单位秒
*/
public void sendMessage(String exchange, String routerKey, String jsonParams,String messageId, int delay) {
log.info("mq发送异步延迟消息:{}", jsonParams);
if (StringUtils.isNotBlank(exchange) && StringUtils.isNotBlank(routerKey)) {
delay = delay * 1000;
MessageProperties props = MessagePropertiesBuilder.newInstance()
.setMessageId(messageId)
.setExpiration(String.valueOf(delay))
.build();
rabbitMqTemplate.convertAndSend(exchange,
routerKey,
MessageBuilder.withBody(jsonParams.getBytes(StandardCharsets.UTF_8)).andProperties(props).build());
} else {
log.warn("参数不完整，未发送MQ消息的，exchange:{},routerKey：{}", exchange, routerKey);
}
}
}
@Resource
private RabbitMqProvider rabbitMqProvider;
rabbitMqProvider.sendMessage(CityMapQueueConfig.CITY_MAP_DEAD_NAME, CityMapQueueConfig.CITY_MAP_ROUTE_DEAD_KEY,JSONObject.toJSONString(callTmpTaskDTO),callTmpTaskDTO.getCityMapTaskDetail().getTaskItemNo(),60);
直接在消息里面设置一个请求头就好了
MessagePropertiesBuilder.newInstance()
.setMessageId(messageId)
.setHeader("delay", l + "")
.build();