Rabbitmq死信取消超时订单

本文使用的版本 otp_win64_25.0rabbitmq-server-3.11.26rabbitmq插件 rabbitmq_delayed_message_exchange-3.11.1

pom.xml文件

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.11.0</version> </dependency>

OrderTimeOutCancelConfig.java

@Configuration @Data public class OrderTimeOutCancelConfig { /** * 正常订单队列，此队列不要绑定消费者，否自无法进入死信队列 */ @Bean public Queue orderQueue() { Map<String, Object> arguments = new HashMap<>(); arguments.put("x-message-ttl", 10000);//超时关闭时间10s arguments.put("x-dead-letter-exchange", RabbitConstants.ORDER_DLX_EXCHANGE); arguments.put("x-dead-letter-routing-key", RabbitConstants.ORDER_DLX_ROUTING_KEY); Queue queue = new Queue(RabbitConstants.ORDER_QUEUE, true, false, false, arguments); return queue; } /** * 自定义交换机 */ @Bean public CustomExchange orderExchange() { Map<String, Object> arguments = new HashMap<>(); arguments.put("x-delayed-type", ExchangeTypes.TOPIC); return new CustomExchange(RabbitConstants.ORDER_EXCHANGE, "x-delayed-message", true, false, arguments); } /** * 绑定交换机和队列 */ @Bean public Binding orderRouting() { return BindingBuilder .bind(orderQueue()) .to(orderExchange()) .with(RabbitConstants.ORDER_ROUTING_KEY) .noargs(); } /** * 订单死信队列 * @return {@link Queue} */ @Bean public Queue orderDlxQueue() { Queue queue = new Queue(RabbitConstants.ORDER_DLX_QUEUE, true, false, false); return queue; } /** * 死信交换机 */ @Bean public Exchange orderDlxExchange() { return new DirectExchange(RabbitConstants.ORDER_DLX_EXCHANGE, true, false, null); } /** * 绑定 */ @Bean public Binding orderDlxRouting() { return BindingBuilder .bind(orderDlxQueue()) .to(orderDlxExchange()) .with(RabbitConstants.ORDER_DLX_ROUTING_KEY) .noargs(); } } @Component @Slf4j public class OrderCancelHandler { /** * 监听订单死信队列，关闭订单 */ @RabbitListener(queues = RabbitConstants.ORDER_DLX_QUEUE, ackMode = "MANUAL") public void onMessage(Message message, Channel channel) throws IOException { log.info("消息进入死信队列..."); String thdTransFlow = new String(message.getBody()); log.info(thdTransFlow + "订单取消"); // 手动ack channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } }

OrderService.java

@Resource

private AmqpTemplate amqpTemplate; public R purchase(){ //其它业务处理 //放入队列 try { amqpTemplate.convertAndSend(RabbitConstants.ORDER_EXCHANGE, RabbitConstants.ORDER_ROUTING_KEY, thdTransFlow.getBytes()); //使用上面默认时间 // 放入死信队列 需要指定其它过期时间时，根据需求二选一即可 //Integer delayTime = 10 * 1000; //amqpTemplate.convertAndSend(RabbitConstants.ORDER_EXCHANGE, RabbitConstants.ORDER_ROUTING_KEY, thdTransFlow, message -> { //message.getMessageProperties().setDelay(delayTime); //return message; //}); } catch (AmqpException e) { log.error("放入取消队列出现异常{}",e.getMessage()); //throw new RuntimeException(e); } //其它业务处理 }