1
spring:
2
  rabbitmq:
3
    host: localhost # 服务端地址（生产环境替换为实际IP）
4
    port: 5672 # 默认端口
5
    username: admin # 账号（Docker安装为admin，原生RabbitMQ默认guest）
6
    password: admin # 密码（与账号对应）
7
    virtual-host: / # 虚拟主机（资源隔离，默认/）
8
    connection-timeout: 15000 # 连接超时时间（毫秒）
9
    # 生产者可靠性配置
10
    publisher-confirm-type: correlated # 异步确认模式，触发ConfirmCallback
11
    publisher-returns: true # 开启消息返回机制，路由失败触发ReturnsCallback
12
    # 消费者可靠性配置
13
    listener:
14
      simple:
15
        acknowledge-mode: manual # 手动ACK，确保业务处理完成后确认消息
16
        prefetch: 10 # 限流：每次最多拉取10条消息，避免消费者过载
17
        concurrency: 1 # 消费者最小线程数
18
        max-concurrency: 5 # 消费者最大线程数

1
import com.hnu.springbootlearning.mq.config.RabbitMQConfig;
2
import jakarta.annotation.Resource;
3
import lombok.Data;
4
import org.springframework.amqp.rabbit.connection.CorrelationData;
5
import org.springframework.amqp.rabbit.core.RabbitTemplate;
6
import org.springframework.stereotype.Component;
7

8
import java.util.UUID;
9

10
/**
11
 * 通用生产者：支持字符串/对象消息发送
12
 */
13
@Component
14
public class RabbitMQProducer {
15
    @Resource
16
    private RabbitTemplate rabbitTemplate;
17

18
    /**
19
     * 发送字符串消息
20
     * @param message 消息内容
21
     */
22
    public void sendSimpleMessage(String message) {
23
        // CorrelationData：消息唯一标识，用于关联发送与确认结果
24
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
25
        rabbitTemplate.convertAndSend(
26
                RabbitMQConfig.BUSINESS_EXCHANGE_NAME,
27
                RabbitMQConfig.BUSINESS_ROUTING_KEY,
28
                message,
29
                correlationData
30
        );
31
    }
32

33
    /**
34
     * 重载：发送字符串消息（指定路由键，测试模糊路由）
35
     * @param message 消息内容
36
     * @param routingKey 路由键
37
     */
38
    public void sendSimpleMessage(String message, String routingKey) {
39
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
40
        rabbitTemplate.convertAndSend(
41
                RabbitMQConfig.BUSINESS_EXCHANGE_NAME,
42
                routingKey,
43
                message,
44
                correlationData
45
        );
46
    }
47

48
    /**
49
     * 发送对象消息（自动序列化为JSON）
50
     * @param user 待发送对象（需符合JavaBean规范）
51
     */
52
    public void sendObjectMessage(User user) {
53
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
54
        rabbitTemplate.convertAndSend(
55
                RabbitMQConfig.BUSINESS_EXCHANGE_NAME,
56
                RabbitMQConfig.BUSINESS_ROUTING_KEY,
57
                user,
58
                correlationData
59
        );
60
    }
61

62
    /**
63
     * 示例：用户实体类（Lombok简化getter/setter）
64
     */
65
    @Data
66
    public static class User {
67
        private Long id;
68
        private String name;
69
        // 可扩展其他字段：age、email等
70
    }
71
}

1
import com.hnu.springbootlearning.mq.config.RabbitMQConfig;
2
import com.rabbitmq.client.Channel;
3
import org.springframework.amqp.core.Message;
4
import org.springframework.amqp.rabbit.annotation.RabbitListener;
5
import org.springframework.stereotype.Component;
6

7
import java.io.IOException;
8
import java.nio.charset.StandardCharsets;
9

10
/**
11
 * 消费者：监听业务队列+死信队列，手动ACK确保消息可靠消费
12
 */
13
@Component
14
public class RabbitMQConsumer {
15

16
    /**
17
     * 监听业务队列（兼容字符串/对象消息）
18
     * @param message 原始消息对象（万能接收，避免类型不匹配）
19
     * @param channel 信道对象（用于手动ACK/NACK）
20
     */
21
    @RabbitListener(queues = RabbitMQConfig.BUSINESS_QUEUE_NAME)
22
    public void consumeBusinessQueue(Message message, Channel channel) throws IOException {
23
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
24
        try {
25
            // 解析消息内容（JSON对象可在此反序列化为JavaBean）
26
            String content = new String(message.getBody(), StandardCharsets.UTF_8);
27
            System.out.println("[业务队列] 接收到消息：" + content);
28

29
            // 模拟业务逻辑（实际开发替换为真实业务：如更新订单、发送通知等）
30
            // 测试死信触发：包含"死信测试"的消息手动抛异常
31
            if (content.contains("死信测试")) {
32
                throw new RuntimeException("模拟消费失败，触发死信队列");
33
            }
34

35
            // 手动ACK：确认消息消费成功，队列删除消息
36
            // multiple=false：仅确认当前消息；true：确认当前及之前所有未确认消息
37
            channel.basicAck(deliveryTag, false);
38
            System.out.println("[业务队列] 消息ACK成功，deliveryTag：" + deliveryTag);
39
        } catch (Exception e) {
40
            System.err.println("[业务队列] 消费失败：" + e.getMessage() + "，deliveryTag：" + deliveryTag);
41
            // 手动NACK：消费失败，拒绝消息并转发到死信队列（requeue=false）
42
            channel.basicNack(deliveryTag, false, false);
43
        }
44
    }
45

46
    /**
47
     * 监听死信队列（处理消费失败/过期消息）
48
     * 死信消息建议：记录日志、持久化、人工干预，一般不重新入队
49
     */
50
    @RabbitListener(queues = RabbitMQConfig.DLX_QUEUE_NAME)
51
    public void consumeDlxQueue(Message message, Channel channel) throws IOException {
52
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
53
        try {
54
            String content = new String(message.getBody(), StandardCharsets.UTF_8);
55
            System.out.println("[死信队列] 接收到消息：" + content);
56

57
            // 死信处理逻辑：如持久化到数据库、发送告警、定时重试等
58
            // handleDlxMessage(content);
59

60
            channel.basicAck(deliveryTag, false);
61
            System.out.println("[死信队列] 消息ACK成功，deliveryTag：" + deliveryTag);
62
        } catch (Exception e) {
63
            System.err.println("[死信队列] 消费失败：" + e.getMessage() + "，deliveryTag：" + deliveryTag);
64
            // 死信消息再次失败：直接ACK删除，避免无限循环
65
            channel.basicAck(deliveryTag, false);
66
        }
67
    }
68

69
    // 示例：死信消息处理方法（实际开发抽离到Service层）
70
    private void handleDlxMessage(String content) {
71
        // 1. 记录死信日志到数据库
72
        // 2. 发送钉钉/邮件告警
73
        // 3. 人工介入处理
74
    }
75
}

1
import com.hnu.springbootlearning.mq.demo.producer.RabbitMQProducer;
2
import jakarta.annotation.Resource;
3
import org.junit.jupiter.api.Test;
4
import org.springframework.boot.test.context.SpringBootTest;
5

6
import java.util.concurrent.TimeUnit;
7

8
/**
9
 * RabbitMQ集成测试：按场景拆分独立测试方法，便于定位问题
10
 * 注解说明：
11
 * - @SpringBootTest：启动Spring容器，加载所有Bean
12
 * - 每个@Test方法独立执行，互不干扰
13
 */
14
@SpringBootTest
15
public class RabbitMQTest {
16

17
    @Resource
18
    private RabbitMQProducer producer;
19

20
    /**
21
     * 测试1：基础字符串消息发送（固定路由键）
22
     * 验证：消息正常发送→到达交换机→业务队列消费→ACK
23
     */
24
    @Test
25
    public void testBasicStringMessage() throws InterruptedException {
26
        // 发送固定路由键的字符串消息
27
        producer.sendSimpleMessage("Hello RabbitMQ!");
28
        // 延迟等待异步回调和消费完成（每个测试独立延迟，避免依赖）
29
        TimeUnit.SECONDS.sleep(1);
30
    }
31

32
    /**
33
     * 测试2：Topic交换机模糊路由
34
     * 验证：business.# 能匹配 business.user.login 等多级路由键
35
     */
36
    @Test
37
    public void testTopicFuzzyRouting() throws InterruptedException {
38
        // 发送多级路由键消息，验证模糊匹配
39
        producer.sendSimpleMessage("测试Topic模糊路由", "business.user.login");
40
        TimeUnit.SECONDS.sleep(1);
41
    }
42

43
    /**
44
     * 测试3：消费失败触发死信队列
45
     * 验证：业务队列消费失败→NACK→消息转发到死信队列→死信队列消费
46
     */
47
    @Test
48
    public void testDlxQueueTrigger() throws InterruptedException {
49
        // 发送含"死信测试"的消息，消费者会模拟异常触发死信
50
        producer.sendSimpleMessage("死信测试：消费失败触发", "business.dlx.test");
51
        TimeUnit.SECONDS.sleep(1);
52
    }
53

54
    /**
55
     * 测试4：对象消息JSON序列化
56
     * 验证：Java对象自动序列化为JSON→消息传输→消费者正常解析
57
     */
58
    @Test
59
    public void testObjectMessageJsonSerialize() throws InterruptedException {
60
        // 构建用户对象并发送
61
        RabbitMQProducer.User user = new RabbitMQProducer.User();
62
        user.setId(1L);
63
        user.setName("张三");
64
        producer.sendObjectMessage(user);
65
        TimeUnit.SECONDS.sleep(1);
66
    }
67

68
    /**
69
     * 测试5：路由失败触发ReturnsCallback
70
     * 验证：消息到达交换机但路由键不匹配→触发Returns回调
71
     */
72
    @Test
73
    public void testRoutingFailReturnCallback() throws InterruptedException {
74
        // 发送不存在的路由键，触发路由失败
75
        producer.sendSimpleMessage("测试路由失败", "error.routing.key");
76
        TimeUnit.SECONDS.sleep(1);
77
    }
78

79
    /**
80
     * 可选：全场景批量测试（保留原有逻辑，便于一次性验证所有场景）
81
     */
82
    @Test
83
    public void testAllScenarios() throws InterruptedException {
84
        testBasicStringMessage();
85
        testTopicFuzzyRouting();
86
        testDlxQueueTrigger();
87
        testObjectMessageJsonSerialize();
88
        testRoutingFailReturnCallback();
89
        TimeUnit.SECONDS.sleep(1);
90
    }
91
}