springboot rabbitmq 延时队列消息确认收货订单已完成

供应商后台-点击发货-默认3天自动收货确认，更新订单状态已完成。

1 pom.xml 引入依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

2 运营后台-订单-发货按钮：生产者发布延时消息

// 5. 发送自动确认收货消息// 计算延迟时间String autoDeliveryDays = configService.getConfigVal(SettingsEnum.AUTO_DELIVERY_DAYS);if (StrUtil.isNotBlank(autoDeliveryDays)) {// long delayTime = Long.parseLong(autoDeliveryDays) * 24 * 60 * 60 * 1000;BigDecimal delayTime = new BigDecimal(autoDeliveryDays).multiply(new BigDecimal(24 * 60 * 60 * 1000));rabbitTemplate.convertAndSend(RabbitMQConfig.DELAY_EXCHANGE,RabbitMQConfig.ORDER_CONFIRM_RECEIPT_ROUTING_KEY,order.getOrderId(),message -> {message.getMessageProperties().setHeader("x-delay", delayTime.longValue());return message;});}

3 RabbitMQ消息队列，路由键，交换机配置

package com.tigshop.common.config;import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;@Configuration
public class RabbitMQConfig { // 路由键public static final String ORDER_CONFIRM_RECEIPT_ROUTING_KEY = "order.confirm.receipt.routing.key";// 队列名称public static final String ORDER_CONFIRM_RECEIPT_QUEUE = "orderConfirmReceiptQueue";/*** 直连交换机（普通交换机）*/@Beanpublic DirectExchange directExchange() {return new DirectExchange(DIRECT_EXCHANGE);}/*** 延迟交换机*/@Beanpublic CustomExchange delayExchange() {Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");return new CustomExchange(DELAY_EXCHANGE, "x-delayed-message", true, false, args);}@Beanpublic Queue orderConfirmReceiptQueue() {return QueueBuilder.durable(ORDER_CONFIRM_RECEIPT_QUEUE).build();}@Beanpublic Binding orderConfirmReceiptBinding() {return BindingBuilder.bind(orderConfirmReceiptQueue()).to(delayExchange()).with(ORDER_CONFIRM_RECEIPT_ROUTING_KEY).noargs();}}

4 消费者实现监听器消费

@RequiredArgsConstructor
@Service
@Slf4j
public class RabbitMqConsumer{@RabbitListener(queues = RabbitMQConfig.ORDER_CONFIRM_RECEIPT_QUEUE)public void receiveOrderConfirmReceiptMessage(Integer orderId) {log.info("收到订单自动确认收货消息：{}", orderId);// 判断是否已经收货Long receivedCount = orderService.lambdaQuery().eq(Order::getOrderId, orderId).eq(Order::getShippingStatus, ShippingStatusEnum.SHIPPED.getCode()).count();if (receivedCount == 1) {return;}// 判断订单是否售后中Long aftersalesCount = aftersalesService.lambdaQuery().eq(Aftersales::getOrderId, orderId).eq(Aftersales::getStatus, AftersalesStatusEnum.IN_REVIEW.getCode()).count();if (aftersalesCount == 1) {return;}try {orderService.confirmReceipt(orderId);} catch (GlobalException e) {log.error("【异常】收到订单自动确认收货消息 RabbitMQ：{}", e.getMessage());}}
}