RabiitMq消息延迟队列使用说明
目前常见的应用软件都有消息的延迟推送的影子，应用也极为广泛，例如：
- 淘宝七天自动确认收货。在我们签收商品后，物流系统会在七天后延时发送一个消息给支付系统，通知支付系统将款打给商家，这个过程持续七天，就是使用了消息中间件的延迟推送功能。
- 12306 购票支付确认页面。我们在选好票点击确定跳转的页面中往往都会有倒计时，代表着 30 分钟内订单不确认的话将会自动取消订单。其实在下订单那一刻开始购票业务系统就会发送一个延时消息给订单系统，延时30分钟，告诉订单系统订单未完成，如果我们在30分钟内完成了订单，则可以通过逻辑代码判断来忽略掉收到的消息
这里我们公司用的技术也是基于RabiitMq来支出订单超时功能
1.安装RabbitMq
这里不会的小伙伴可参考我的另一篇博文，RabiitMq的安装
2.代码绑定交换机
订单交换机队列：直接上代码
详细代码如下：
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @description: 订单延迟处理直连交换机
* @createDate: 2020/5/26
*/
@Configuration
public class DirectExchangeConfig {
/**
* 订单无消费者交换机
*/
public static final String ORDER_DIRECT_EXCHANGE = "ORDER_DIRECT_EXCHANGE";
/**
* 订单无消费者队列
*/
public static final String ORDER_DIRECT_QUEUE = "ORDER_DIRECT_QUEUE";
/**
* 直连路由KEY
*/
public static final String ORDER_DIRECT_ROUTEKEY = "ORDER_DELAY_KEY";
@Bean
public DirectExchange orderDirectExchange(){
//指定消息队列参数，目前可以设置为空
HashMap argument = new HashMap();
return new DirectExchange(ORDER_DIRECT_EXCHANGE,true,false,null);
}
/**
* 订单延迟队列，此队列不指定消费者,设置过期时间
* @return
*/
@Bean
public Queue orderDirectQueue(){
//这里只需要设置持久化开启即可，注意队列的持久化和交换机的持久化都要设置，否则重启mq会导致消息丢失
HashMap argument = new HashMap();
argument.put("x-dead-letter-exchange",ORDER_DEAD_EXCHANGE);
argument.put("x-dead-letter-routing-key",ORDER_DEAD_ROUTEKEY);
return new Queue(ORDER_DIRECT_QUEUE,true,false,false,argument);
}
/**
* 绑定队列
*/
@Bean
Binding bindingDirect(){
return BindingBuilder.bind(orderDirectQueue()).to(orderDirectExchange()).with(ORDER_DIRECT_ROUTEKEY);
}
}
死信交换机及队列：
package com.thirtydays.payment.mq.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @description: 死信队列相关配置信息
* @createDate: 2020/5/26
*/
@Configuration
public class RabbitDeadConfig {
/**
* 死信队列交换机名称
*/
public static final String ORDER_DEAD_EXCHANGE = "ORDER_DEAD_EXCHANGE";
/**
* 死信队列名称
*/
public static final String ORDER_DEAD_QUEUE = "ORDER_DEAD_QUEUE";
/**
* 死信队列路由key
*/
public static final String ORDER_DEAD_ROUTEKEY = "ORDER_DEAD_ROUTEKEY";
/*
* 定义一个死信队列绑定的交换机
*/
@Bean
public DirectExchange deadExchange(){
//指定消息队列参数，目前可以设置为空
HashMap argument = new HashMap();
return new DirectExchange(ORDER_DEAD_EXCHANGE,true,false,null);
}
/**
* 定义一个死信队列
*/
@Bean
public Queue deadQueue(){
return new Queue(ORDER_DEAD_QUEUE,true);
}
/**
* 绑定死信队列和交换机
*/
@Bean
public Binding bindingDeadQueue(){
return BindingBuilder.bind(deadQueue()).to(deadExchange()).with(ORDER_DEAD_ROUTEKEY);
}
}
3.自定义配置RabiitMq线程数
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Map;
/**
* @description: RabbitMq相关配置
* @createDate: 2020/5/28
*/
@Configuration
@Slf4j
public class RabbitMqConfig implements RabbitTemplate.ConfirmCallback{
@Autowired
public RabbitMqConfig(RabbitTemplate rabbitTemplate){
rabbitTemplate.setConfirmCallback(this);
}
/**
* 配置mq自定义工厂 可以查看MQ默认配置不支持多线程，这里配置使用多线程可以防止消费者消息积压
* 后续可以看实际效果的容错能力来进行线程数的调整和优惠
* @param configurer
* @param connectionFactory
* @return
*/
@Bean("customContainerFactory")
public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
//设置线程数
factory.setConcurrentConsumers(5);
//最大线程数
factory.setMaxConcurrentConsumers(5);
configurer.configure(factory, connectionFactory);
return factory;
}
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if(ack){
log.info("消息已成功发送到交换机=====>");
}
else {
log.info("消息发送到交换机失败:{}=====>{}",cause);
}
}
}
4.配置消息生产者 和 消息监听器
生产者
import com.alibaba.fastjson.JSON;
import com.thirtydays.payment.model.bo.OrderBaseBO;
import com.thirtydays.payment.model.bo.OrderGroupBaseBO;
import com.thirtydays.payment.mq.config.DirectExchangeConfig;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.stereotype.Component;
import java.util.UUID;
/**
* @description:
* @createDate: 2020/5/26
*/
@Component
@AllArgsConstructor
@Slf4j
public class MqProducer {
private RabbitTemplate rabbitTemplate;
public void sendOrderMessageQueue(OrderBaseBO orderBaseBO){
String orderMessage = JSON.toJSONString(orderBaseBO);
//声明消息,在此处声明消息的过期时间，到期时间即进入死信队列中去
MessagePostProcessor postProcessor = message -> {
MessageProperties messageProperties = message.getMessageProperties();
messageProperties.setContentEncoding("utf-8");
messageProperties.setMessageId(orderBaseBO.getOrderNo());
messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
//设置过期时间为半个小时（1800S）加5s中，主要是为了方式微信拉起支付延迟会导致微信支付成功而导致mq订单取消
messageProperties.setExpiration(String.valueOf(1000*1805));
return message;
};
rabbitTemplate.convertAndSend(DirectExchangeConfig.ORDER_DIRECT_EXCHANGE,DirectExchangeConfig.ORDER_DIRECT_ROUTEKEY,orderMessage,postProcessor);
log.info("成功发送订单信息到消息队列中,订单信息========>:{}",orderBaseBO);
}
}
消息监听
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
/**
* @description:
* @createDate: 2020/5/26
*/
@Component
@Slf4j
@AllArgsConstructor
public class MqOrderListener {
private OrderService orderService;
/**
* 超时取消订单
* @param message
* @param channel
* @throws IOException
*/
@RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = RabbitDeadConfig.ORDER_DEAD_EXCHANGE,autoDelete = "false",durable = "true"),
value = @Queue(value = RabbitDeadConfig.ORDER_DEAD_QUEUE,autoDelete = "false",durable = "true"),
key = RabbitDeadConfig.ORDER_DEAD_ROUTEKEY),containerFactory = "customContainerFactory")
public void orderListener(Message message,Channel channel) throws IOException {
//取消订单
String orderMessage = new String(message.getBody());
try {
OrderBaseBO orderBaseBO = JSON.parseObject(orderMessage, OrderBaseBO.class);
log.info("监听到延时订单消息===========》订单内容:{}",orderBaseBO);
//此处补充取消订单的逻辑 一般将订单状态改成超时等操作
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}catch (Exception e){
log.error("取消订单出现异常:{}",e);
channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
//此处可记录异常信息到日志表中
}
}
}