本文使用的版本
- otp_win64_25.0
- rabbitmq-server-3.11.26
- rabbitmq插件 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 {/*** 正常订单队列，此队列不要绑定消费者，否自无法进入死信队列*/@Beanpublic Queue orderQueue() {Map<String, Object> arguments = new HashMap<>();arguments.put("x-message-ttl", 10000);//超时关闭时间10sarguments.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;}/*** 自定义交换机*/@Beanpublic 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);}/*** 绑定交换机和队列*/@Beanpublic Binding orderRouting() {return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(RabbitConstants.ORDER_ROUTING_KEY).noargs();}/*** 订单死信队列* @return {@link Queue}*/@Beanpublic Queue orderDlxQueue() {Queue queue = new Queue(RabbitConstants.ORDER_DLX_QUEUE, true, false, false);return queue;}/*** 死信交换机*/@Beanpublic Exchange orderDlxExchange() {return new DirectExchange(RabbitConstants.ORDER_DLX_EXCHANGE, true, false, null);}/*** 绑定*/@Beanpublic 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 + "订单取消");// 手动ackchannel.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);} //其它业务处理}
)
