1
@Test
2
public void testPublisherConfirm(){
3
    CorrelationData correlationData = new CorrelationData();
4

5
    // 添加回调处理
6
    correlationData.getFuture()
7
        .thenAccept(confirm -> {
8
            if(confirm.isAck()){
9
                log.info("消息发送成功,收到Ack");
10
            }else{
11
                log.info("消息发送失败,原因 {},id {}",confirm.getReason(),correlationData.getId());
12
            }
13
        })
14
        .exceptionally(throwable -> {
15
            log.error("消息发送失败,原因 {}",throwable.getMessage());
16
            return null;
17
        });
18

19
    rabbitTemplate.convertAndSend("cx.direct", "yell", "hello,confirm", correlationData);
20
}

1
@Bean
2
public Queue lazyQueue(){
3
    return QueueBuilder
4
            .durable("lazy.queue")
5
            .lazy() // 开启Lazy模式
6
            .build();
7
}

1
@RabbitListener(queuesToDeclare = @Queue(
2
        name = "lazy.queue",
3
        durable = "true",
4
        arguments = @Argument(name = "x-queue-mode", value = "lazy")
5
))
6
public void listenLazyQueue(String msg){
7
    log.info("接收到 lazy.queue的消息：{}", msg);
8
}

1
rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues

1
try {
2
    // 业务逻辑处理
3
    processMessage(message);
4
    return ack;  // 成功时自动返回ack
5
} catch (Exception e) {
6
    if (isBusinessException(e)) {
7
        return nack;  // 业务异常，返回nack重试
8
    } else {
9
        return reject;  // 消息格式异常，返回reject拒绝
10
    }
11
}

1
@RabbitListener(queues = "simple.queue")
2
public void listenSimpleQueueMessage(String msg) {
3
    log.info("消费者接收到消息：【{}】", msg);
4
    if (true) {
5
        throw new MessageConversionException("故意的异常");
6
    }
7
    log.info("消息处理完成");
8
}

1
spring:
2
  rabbitmq:
3
    listener:
4
      simple:
5
        acknowledge-mode: none

1
spring:
2
  rabbitmq:
3
    listener:
4
      simple:
5
        acknowledge-mode: auto

1
@RabbitListener(queues = "simple.queue")
2
public void listenSimpleQueueMessage(String msg) {
3
    log.info("消费者接收到消息：【{}】", msg);
4
    if (true) {
5
        throw new MessageConversionException("消息转换异常");  // 会返回reject
6
    }
7
}

1
@RabbitListener(queues = "simple.queue")
2
public void listenSimpleQueueMessage(String msg) {
3
    log.info("消费者接收到消息：【{}】", msg);
4
    if (true) {
5
        throw new RuntimeException("业务异常");  // 会返回nack
6
    }
7
}

1
package com.itheima.consumer.config;
2

3
import org.springframework.amqp.core.*;
4
import org.springframework.amqp.rabbit.core.RabbitTemplate;
5
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
6
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
7
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
8
import org.springframework.context.annotation.Bean;
9
import org.springframework.context.annotation.Configuration;
10

11
@Configuration
12
@ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
13
public class ErrorMessageConfig {
14

15
    /**
16
     * 定义异常交换机
17
     */
18
    @Bean
19
    public DirectExchange errorMessageExchange(){
20
        return new DirectExchange("error.direct");
21
    }
22

23
    /**
24
     * 定义异常队列
25
     */
26
    @Bean
27
    public Queue errorQueue(){
28
        return new Queue("error.queue", true);
29
    }
30

31
    /**
32
     * 绑定异常队列到异常交换机
33
     */
34
    @Bean
35
    public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
36
        return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
37
    }
38

39
    /**
40
     * 定义RepublishMessageRecoverer，关联队列和交换机
41
     */
42
    @Bean
43
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
44
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
45
    }
46
}

1
@RabbitListener(queues = "error.queue")
2
public void handleErrorMessage(String msg,
3
                              @Header Map<String, Object> headers,
4
                              @Header("x-exception-message") String exceptionMessage) {
5
    log.error("收到异常消息：{}", msg);
6
    log.error("异常信息：{}", exceptionMessage);
7
    log.error("原始headers：{}", headers);
8

9
    // 1. 记录到数据库
10
    // 2. 发送告警通知
11
    // 3. 尝试修复后重新发送
12
}

1
@Bean
2
public MessageConverter messageConverter(){
3
    Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
4
    // 配置自动创建消息ID，用于识别不同消息
5
    converter.setCreateMessageIds(true);
6
    return converter;
7
}

1
@RabbitListener(queues = "payment.queue")
2
public void handlePaymentMessage(PaymentMessage msg,
3
                                @Header("spring_returned_message_correlation") String messageId) {
4
    // 1. 检查消息ID是否已处理
5
    if (messageIdService.isProcessed(messageId)) {
6
        log.info("消息已处理，跳过：{}", messageId);
7
        return;
8
    }
9

10
    // 2. 处理业务逻辑
11
    processPayment(msg);
12

13
    // 3. 记录消息ID
14
    messageIdService.markProcessed(messageId);
15
}

1
@Service
2
public class MessageIdService {
3
    @Autowired
4
    private RedisTemplate<String, String> redisTemplate;
5

6
    private static final String PROCESSED_KEY_PREFIX = "msg:processed:";
7
    private static final long EXPIRE_TIME = 24 * 60 * 60; // 24小时过期
8

9
    public boolean isProcessed(String messageId) {
10
        String key = PROCESSED_KEY_PREFIX + messageId;
11
        return redisTemplate.hasKey(key);
12
    }
13

14
    public void markProcessed(String messageId) {
15
        String key = PROCESSED_KEY_PREFIX + messageId;
16
        redisTemplate.opsForValue().set(key, "1", EXPIRE_TIME, TimeUnit.SECONDS);
17
    }
18
}

1
@Override
2
public void markOrderPaySuccess(Long orderId) {
3
    // 方式1：先查询再更新（存在并发问题）
4
    Order old = getById(orderId);
5
    if (old == null || old.getStatus() != 1) {
6
        // 订单不存在或状态不是未支付，跳过处理
7
        return;
8
    }
9

10
    // 更新订单状态
11
    Order order = new Order();
12
    order.setId(orderId);
13
    order.setStatus(2); // 已支付
14
    order.setPayTime(LocalDateTime.now());
15
    updateById(order);
16
}

1
@Override
2
public void markOrderPaySuccess(Long orderId) {
3
    // 使用WHERE条件确保幂等性
4
    int updated = lambdaUpdate()
5
            .set(Order::getStatus, 2)           // 设置为已支付
6
            .set(Order::getPayTime, LocalDateTime.now())
7
            .eq(Order::getId, orderId)          // 订单ID匹配
8
            .eq(Order::getStatus, 1)            // 只有未支付状态才更新
9
            .update();
10

11
    if (updated == 0) {
12
        log.info("订单{}状态更新跳过，可能已处理", orderId);
13
    }
14
}

1
UPDATE `order`
2
SET status = 2, pay_time = NOW()
3
WHERE id = ? AND status = 1

1
@Component
2
@Slf4j
3
public class OrderStatusCheckTask {
4

5
    @Autowired
6
    private OrderService orderService;
7

8
    @Autowired
9
    private PaymentService paymentService;
10

11
    /**
12
     * 每20秒检查一次未支付订单
13
     */
14
    @Scheduled(fixedDelay = 20000)
15
    public void checkUnpaidOrders() {
16
        // 查询5分钟内创建但未支付的订单
17
        LocalDateTime cutoffTime = LocalDateTime.now().minusMinutes(5);
18
        List<Order> unpaidOrders = orderService.getUnpaidOrdersAfter(cutoffTime);
19

20
        for (Order order : unpaidOrders) {
21
            checkAndUpdateOrderStatus(order);
22
        }
23
    }
24

25
    private void checkAndUpdateOrderStatus(Order order) {
26
        try {
27
            // 查询支付状态
28
            PaymentStatus status = paymentService.getPaymentStatus(order.getId());
29

30
            if (status == PaymentStatus.PAID) {
31
                // 支付成功，更新订单状态
32
                orderService.markOrderPaySuccess(order.getId());
33
                log.info("定时任务检测到订单{}已支付，状态已更新", order.getId());
34
            }
35
        } catch (Exception e) {
36
            log.error("检查订单{}支付状态失败", order.getId(), e);
37
        }
38
    }
39
}

1
@Service
2
public class OrderServiceImpl implements OrderService {
3

4
    /**
5
     * 查询指定时间后创建的未支付订单
6
     */
7
    public List<Order> getUnpaidOrdersAfter(LocalDateTime createTime) {
8
        return lambdaQuery()
9
                .eq(Order::getStatus, 1)  // 未支付状态
10
                .ge(Order::getCreateTime, createTime)  // 创建时间大于指定时间
11
                .list();
12
    }
13
}

1
// 只查询特定时间窗口内的订单
2
LocalDateTime startTime = LocalDateTime.now().minusMinutes(30);  // 30分钟前
3
LocalDateTime endTime = LocalDateTime.now().minusMinutes(1);     // 1分钟前
4

5
List<Order> orders = orderService.getUnpaidOrdersBetween(startTime, endTime);

1
@Scheduled(fixedDelay = 20000)
2
public void checkUnpaidOrdersBatch() {
3
    int pageSize = 100;
4
    int page = 0;
5

6
    while (true) {
7
        List<Order> orders = orderService.getUnpaidOrdersPaged(page, pageSize);
8
        if (orders.isEmpty()) break;
9

10
        for (Order order : orders) {
11
            checkAndUpdateOrderStatus(order);
12
        }
13
        page++;
14
    }
15
}

1
@Async
2
public void checkOrderStatusAsync(Order order) {
3
    checkAndUpdateOrderStatus(order);
4
}

1
// 配置死信交换机
2
@Bean
3
public Queue orderQueue() {
4
    return QueueBuilder.durable("order.queue")
5
            .withArgument("x-dead-letter-exchange", "order.dlx.exchange")    // 死信交换机
6
            .withArgument("x-dead-letter-routing-key", "order.timeout")      // 死信路由键
7
            .withArgument("x-message-ttl", 30 * 60 * 1000)                   // 30分钟TTL
8
            .build();
9
}

1
// 发送消息到ttl.fanout，routingKey为blue，TTL=5000ms
2
rabbitTemplate.convertAndSend("ttl.fanout", "blue", "延迟消息", message -> {
3
    message.getMessageProperties().setExpiration("5000");  // 5秒后过期
4
    return message;
5
});

1
发送者 → 交换机 → [立即路由] → 队列 → 消费者

1
发送者 → 交换机 → [立即路由] → 队列 → [等待TTL过期] → 死信交换机 → 死信队列 → 消费者

1
发送者 → 延迟交换机 → [等待延迟时间] → 路由到队列 → 消费者 [立即消费]

1
# 复制到RabbitMQ的plugins目录
2
cp rabbitmq_delayed_message_exchange-3.10.2.ez /usr/lib/rabbitmq/lib/rabbitmq_server-3.10.2/plugins/

1
# 进入sbin目录
2
cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.10.2/sbin
3

4
# 启用插件
5
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

1
systemctl restart rabbitmq-server

1
@RabbitListener(bindings = @QueueBinding(
2
        value = @Queue(name = "delay.queue", durable = "true"),
3
        exchange = @Exchange(name = "delay.direct", delayed = "true"),  // 关键：delayed = "true"
4
        key = "delay"
5
))
6
public void listenDelayMessage(String msg){
7
    log.info("接收到delay.queue的延迟消息：{}", msg);
8
}

1
@Configuration
2
public class DelayExchangeConfig {
3

4
    /**
5
     * 延迟交换机 - 本质是DirectExchange + 延迟功能
6
     */
7
    @Bean
8
    public DirectExchange delayExchange() {
9
        return ExchangeBuilder.directExchange("delay.exchange")
10
                .durable(true)
11
                .delayed()  // 🎯 关键：启用延迟功能
12
                .build();
13
    }
14

15
    /**
16
     * 普通队列
17
     */
18
    @Bean
19
    public Queue delayQueue() {
20
        return new Queue("delay.queue", true);
21
    }
22

23
    /**
24
     * 绑定队列和延迟交换机
25
     */
26
    @Bean
27
    public Binding delayBinding(Queue delayQueue, DirectExchange delayExchange){
28
        return BindingBuilder.bind(delayQueue)
29
                .to(delayExchange)
30
                .with("delay");  // routing key
31
    }
32
}

1
/**
2
 * 测试延迟消息
3
 */
4
@Test
5
public void testDelayMessage() {
6
    String exchangeName = "delay.exchange";
7
    String routingKey = "delay";
8
    String message = "hello,delay";
9

10
    // 使用MessagePostProcessor设置延迟时间
11
    rabbitTemplate.convertAndSend(exchangeName, routingKey, message, msg -> {
12
        // 🎯 设置延迟时间为10秒（毫秒单位）
13
        msg.getMessageProperties().setHeader("x-delay", 10000);
14
        return msg;
15
    });
16

17
    log.info("延迟消息已发送，将在10秒后被消费");
18
}

1
@Service
2
public class DelayMessageService {
3

4
    @Autowired
5
    private RabbitTemplate rabbitTemplate;
6

7
    /**
8
     * 发送延迟消息
9
     * @param exchange 交换机名称
10
     * @param routingKey 路由键
11
     * @param message 消息内容
12
     * @param delaySeconds 延迟秒数
13
     */
14
    public void sendDelayMessage(String exchange, String routingKey, Object message, int delaySeconds) {
15
        rabbitTemplate.convertAndSend(exchange, routingKey, message, msg -> {
16
            msg.getMessageProperties().setHeader("x-delay", delaySeconds * 1000);
17
            return msg;
18
        });
19
    }
20
}

1
@Service
2
public class OrderService {
3

4
    @Autowired
5
    private DelayMessageService delayMessageService;
6

7
    /**
8
     * 创建订单并发送超时检查消息
9
     */
10
    public void createOrder(Order order) {
11
        // 1. 保存订单
12
        orderMapper.insert(order);
13

14
        // 2. 发送30分钟后的超时检查消息
15
        delayMessageService.sendDelayMessage(
16
            "order.delay.exchange",
17
            "order.timeout.check",
18
            order.getId(),
19
            30 * 60  // 30分钟
20
        );
21

22
        log.info("订单创建成功，30分钟后将检查支付状态");
23
    }
24
}

1
/**
2
 * 分层检查订单状态
3
 */
4
public void sendOrderCheckMessages(Long orderId) {
5
    // 5分钟后第一次检查
6
    delayMessageService.sendDelayMessage("order.delay.exchange", "order.check", orderId, 5 * 60);
7

8
    // 15分钟后第二次检查
9
    delayMessageService.sendDelayMessage("order.delay.exchange", "order.check", orderId, 15 * 60);
10

11
    // 30分钟后最终检查
12
    delayMessageService.sendDelayMessage("order.delay.exchange", "order.timeout", orderId, 30 * 60);
13
}

1
// 不要发送6小时的延迟消息
2
// 而是每小时发送一次检查消息
3
for (int i = 1; i <= 6; i++) {
4
    sendDelayMessage("check.exchange", "hourly.check", orderId, i * 60 * 60);
5
}

1
// 延迟消息 + 定时任务的混合方案
2
// 延迟消息处理短期任务（< 1小时）
3
// 定时任务处理长期任务（> 1小时）

1
@Service
2
public class OrderDelayService {
3

4
    @Autowired
5
    private DelayMessageService delayMessageService;
6

7
    /**
8
     * 订单超时取消场景
9
     */
10
    public void handleOrderTimeout(Long orderId) {
11
        // 发送分段检查消息，避免长时间占用资源
12

13
        // 10分钟后首次检查
14
        delayMessageService.sendDelayMessage(
15
            "order.delay.exchange",
16
            "order.check",
17
            orderId,
18
            10 * 60
19
        );
20

21
        // 20分钟后二次检查
22
        delayMessageService.sendDelayMessage(
23
            "order.delay.exchange",
24
            "order.check",
25
            orderId,
26
            20 * 60
27
        );
28

29
        // 30分钟后最终超时处理
30
        delayMessageService.sendDelayMessage(
31
            "order.delay.exchange",
32
            "order.timeout",
33
            orderId,
34
            30 * 60
35
        );
36
    }
37

38
    /**
39
     * 处理订单检查消息
40
     */
41
    @RabbitListener(queues = "order.check.queue")
42
    public void handleOrderCheck(Long orderId) {
43
        Order order = orderService.getById(orderId);
44

45
        if (order == null || order.getStatus() != OrderStatus.UNPAID) {
46
            log.info("订单{}状态已变更，跳过检查", orderId);
47
            return;
48
        }
49

50
        // 检查支付状态
51
        PaymentStatus paymentStatus = paymentService.getPaymentStatus(orderId);
52
        if (paymentStatus == PaymentStatus.PAID) {
53
            orderService.markOrderPaid(orderId);
54
            log.info("订单{}支付成功", orderId);
55
        }
56
    }
57

58
    /**
59
     * 处理订单超时
60
     */
61
    @RabbitListener(queues = "order.timeout.queue")
62
    public void handleOrderTimeout(Long orderId) {
63
        Order order = orderService.getById(orderId);
64

65
        if (order != null && order.getStatus() == OrderStatus.UNPAID) {
66
            // 取消订单并释放库存
67
            orderService.cancelOrder(orderId);
68
            inventoryService.releaseInventory(orderId);
69
            log.info("订单{}超时取消", orderId);
70
        }
71
    }
72
}

1
@Service
2
public class CouponReminderService {
3

4
    /**
5
     * 优惠券到期提醒
6
     */
7
    public void sendCouponExpiryReminder(Long couponId, LocalDateTime expiryTime) {
8
        LocalDateTime now = LocalDateTime.now();
9

10
        // 过期前3天提醒
11
        LocalDateTime remind3Days = expiryTime.minusDays(3);
12
        if (remind3Days.isAfter(now)) {
13
            long delay3Days = Duration.between(now, remind3Days).getSeconds();
14
            delayMessageService.sendDelayMessage(
15
                "coupon.reminder.exchange",
16
                "coupon.expire.3days",
17
                couponId,
18
                (int) delay3Days
19
            );
20
        }
21

22
        // 过期前1天提醒
23
        LocalDateTime remind1Day = expiryTime.minusDays(1);
24
        if (remind1Day.isAfter(now)) {
25
            long delay1Day = Duration.between(now, remind1Day).getSeconds();
26
            delayMessageService.sendDelayMessage(
27
                "coupon.reminder.exchange",
28
                "coupon.expire.1day",
29
                couponId,
30
                (int) delay1Day
31
            );
32
        }
33
    }
34

35
    @RabbitListener(queues = "coupon.expire.3days.queue")
36
    public void handleCoupon3DaysReminder(Long couponId) {
37
        // 发送3天到期提醒
38
        notificationService.sendCouponExpiryReminder(couponId, 3);
39
    }
40

41
    @RabbitListener(queues = "coupon.expire.1day.queue")
42
    public void handleCoupon1DayReminder(Long couponId) {
43
        // 发送1天到期提醒
44
        notificationService.sendCouponExpiryReminder(couponId, 1);
45
    }
46
}