1
@Test
2
    void testSendMessage() {
3
        // 消息自定义为非持久化
4
        Message message = MessageBuilder.withBody("hello springAMQP".getBytes(StandardCharsets.UTF_8))
5
                .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();
6
        for (int i = 0; i < 1000000; i++) {
7
            rabbitTemplate.convertAndSend("simple.queue", message);
8
        }
9
    }

1
    @Test
2
    void testSendMessage() {
3
        // 消息持久化
4
        Message message = MessageBuilder.withBody("hello springAMQP".getBytes(StandardCharsets.UTF_8))
5
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
6
        for (int i = 0; i < 1000000; i++) {
7
            rabbitTemplate.convertAndSend("simple.queue", message);
8
        }
9
    }

1
spring:
2
  rabbitmq:
3
    listener:
4
      simple:
5
        acknowledge-mode: none # 不做处理  auto # 自动ack   manual 手动ack

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

5
    @RabbitListener(queues = "simple.queue")
6
    public void listenSimpleQueue(String msg) {
7
        log.info("监听到simple.queue的消息:{}", msg);
8
        throw new RuntimeException("故意的");
9
    }
10
}

1
throw new MessageConversionException("故意的");

1
06-03 21:44:58:825  WARN 10276 --- [ntContainer#3-1] ingErrorHandler$DefaultExceptionStrategy : Fatal message conversion error; message rejected; it will be dropped or routed to a dead letter exchange, if so configured: (Body:'"Hello Spring AMQP!"' MessageProperties [headers={__TypeId__=java.lang.String}, messageId=eb2e4bd1-6077-4419-be95-90ebfb307fae, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=true, receivedExchange=, receivedRoutingKey=simple.queue, deliveryTag=1, consumerTag=amq.ctag-tZhHXR8ze87Bjd1Ye_1Tww, consumerQueue=simple.queue])
2
06-03 21:44:58:825 ERROR 10276 --- [ntContainer#3-1] o.s.a.r.l.SimpleMessageListenerContainer : Execution of Rabbit message listener failed, and the error handler threw an exception

1
@Bean
2
    public MessageConverter messageConverter(){
3
        // 1.定义消息转换器
4
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
5
        // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
6
        jackson2JsonMessageConverter.setCreateMessageIds(true);
7
        return jackson2JsonMessageConverter;
8
    }

1
    @RabbitListener(queues = "simple.queue")
2
    public void listenSimpleQueue(Message message) {
3
        log.info("监听到simple.queue的ID:{}", message.getMessageProperties().getMessageId());
4
        log.info("监听到simple.queue的消息:{}", new String(message.getBody()));
5
    }

1
06-04 21:00:06:868  INFO 3344 --- [ntContainer#3-1] c.i.consumer.mq.SpringRabbitListener     : 监听到simple.queue的ID:480f70f0-d658-4447-824b-022c3b8970e3
2
06-04 21:00:06:869  INFO 3344 --- [ntContainer#3-1] c.i.consumer.mq.SpringRabbitListener     : 监听到simple.queue的消息:"Hello Spring AMQP!"

1
@Component
2
@RequiredArgsConstructor
3
public class PayStatusListener {
4

5
    private final IOrderService orderService;
6

7
    @RabbitListener(bindings = @QueueBinding(
8
            value = @Queue(name = "trade.pay.success.queue", durable = "true"),
9
            exchange = @Exchange(name = "pay.direct"),
10
            key = "pay.success"
11
    ))
12
    public void listenPaySuccess(Long orderId) {
13
        Order order = orderService.getById(orderId);
14
        // 判断订单状态，是否为未支付
15
        if (order == null || order.getStatus() != 1) {
16
            // 不做处理
17
            return;
18
        }
19
        // 标记订单状态为已支付
20
        orderService.markOrderPaySuccess(orderId);
21
    }
22
}

1
    @RabbitListener(bindings = @QueueBinding(
2
            value = @Queue("dlx.queue"),
3
            exchange = @Exchange(name = "dlx.direct", type = ExchangeTypes.DIRECT),
4
            key = {"hi"}
5
    ))
6
    public void listenDlxQueue(String msg) {
7
        log.info("消费者2监听到dlx.queue的消息:{}", msg);
8
    }

1
@Configuration
2
public class NormalConfig {
3

4
    @Bean
5
    public DirectExchange normalExchange() {
6
        return ExchangeBuilder.directExchange("normal.direct").build();
7
    }
8

9
    @Bean
10
    public Queue normalQueue() {
11
        return QueueBuilder
12
                .durable("normal.queue")
13
                .deadLetterExchange("dlx.direct")  // 指定死信交换机
14
                .build();
15
    }
16

17
    @Bean
18
    public Binding normalExchangeBinding(Queue normalQueue, DirectExchange normalExchange) {
19
        return BindingBuilder.bind(normalQueue).to(normalExchange).with("hi");
20
    }
21

22
}

1
    @Test
2
    void testSendDelayMessage() {
3
        rabbitTemplate.convertAndSend("normal.direct", "hi", "hello lalala", message -> {
4
            message.getMessageProperties().setExpiration("10000");
5
            return message;
6
        });
7
    }

1
[
2
    {
3
        "CreatedAt": "2024-06-19T09:22:59+08:00",
4
        "Driver": "local",
5
        "Labels": null,
6
        "Mountpoint": "/var/lib/docker/volumes/mq-plugins/_data",
7
        "Name": "mq-plugins",
8
        "Options": null,
9
        "Scope": "local"
10
    }
11
]

1
@RabbitListener(bindings = @QueueBinding(
2
            value = @Queue("delay.queue"),
3
            exchange = @Exchange(name = "delay.direct", delayed = "true", type = ExchangeTypes.DIRECT),
4
            key = {"hi"}
5
    ))
6
    public void listenDelayQueue(String msg) {
7
        log.info("消费者监听到delay.queue的消息:{}", msg);
8
    }

1
    @Test
2
    void testSendDelayMessageByPlugins() {
3
        rabbitTemplate.convertAndSend("delay.direct", "hi", "hello lalala", message -> {
4
            message.getMessageProperties().setDelay(10000);
5
            return message;
6
        });
7
    }

1
public interface MQConstans {
2
    String DELAY_EXCHANGE_NAME = "trade.delay.direct";
3
    String DELAY_ORDER_QUEUE_NAME = "trade.delay.order.queue";
4
    String DELAY_ORDER_KEY_NAME = "delay.order.query";
5
}

1
@Component
2
@RequiredArgsConstructor
3
public class OrderDelayMessageListener {
4

5
    private final IOrderService orderService;
6
    private final PayClient payClient;
7

8
    @RabbitListener(bindings = @QueueBinding(
9
            value = @Queue(name = MQConstans.DELAY_ORDER_QUEUE_NAME),
10
            exchange = @Exchange(name = MQConstans.DELAY_EXCHANGE_NAME, delayed = "true"),
11
            key = MQConstans.DELAY_ORDER_KEY_NAME
12
    ))
13
    public void listenOrderDelayMessage(Long orderId) {
14
        // 1.查询订单
15
        Order order = orderService.getById(orderId);
16
        // 2.检测订单状态，判断是否己支付
17
        if (order == null && order.getStatus() != 1) {
18
            // 订单不存在 或 已支付
19
            return;
20
        }
21
        // 3. 未支付，需要查询支付流水状态
22
        PayOrderDTO payOrderDTO = payClient.queryPayOrderByBizOrderNo(orderId);
23
        // 4.判断是否支付
24
        if (payOrderDTO != null && payOrderDTO.getStatus() == 3) {
25
            // 4.1.已支付，标记订单状态为已支付
26
            orderService.markOrderPaySuccess(orderId);
27
        } else {
28
            // 4.2.未支付，取消订单，回复库存
29
            orderService.cancelOrder(orderId);
30
        }
31
    }
32
}