引言

本文讲解RabbitMQ的高级特性和生产环境最佳实践。

一、死信队列（DLX）

1
2
3
4
5
6
7
8
9
10
11
// 声明死信交换机
channel.exchangeDeclare("dlx_exchange", "direct");
channel.queueDeclare("dlx_queue", true, false, false, null);
channel.queueBind("dlx_queue", "dlx_exchange", "dlx_routing_key");

// 业务队列绑定死信交换机
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx_exchange");
args.put("x-dead-letter-routing-key", "dlx_routing_key");
args.put("x-message-ttl", 10000); // 10秒过期
channel.queueDeclare("business_queue", true, false, false, args);

死信场景：

  1. 消息被拒绝（basic.reject/basic.nack）且requeue=false
  2. 消息TTL过期
  3. 队列达到最大长度

二、延迟队列

2.1 使用TTL+DLX实现

1
2
3
4
5
// 延迟30秒
Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 30000);
args.put("x-dead-letter-exchange", "delay_exchange");
channel.queueDeclare("delay_queue", true, false, false, args);

2.2 使用延迟插件

1
2
3
4
5
6
7
8
9
10
// 安装rabbitmq_delayed_message_exchange插件
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare("delayed_exchange", "x-delayed-message", true, false, args);

// 发送延迟消息
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .headers(Collections.singletonMap("x-delay", 30000))
    .build();
channel.basicPublish("delayed_exchange", "routing_key", props, message.getBytes());

三、优先级队列

1
2
3
4
5
6
7
8
9
10
// 声明优先级队列
Map<String, Object> args = new HashMap<>();
args.put("x-max-priority", 10);
channel.queueDeclare("priority_queue", true, false, false, args);

// 发送带优先级的消息
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .priority(5)
    .build();
channel.basicPublish("", "priority_queue", props, message.getBytes());

四、消息幂等性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 使用消息ID去重
@Component
public class MessageConsumer {
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @RabbitListener(queues = "order_queue")
    public void handleMessage(Message message) {
        String messageId = message.getMessageProperties().getMessageId();
        
        // 检查是否已处理
        String key = "msg:" + messageId;
        if (redisTemplate.opsForValue().setIfAbsent(key, "1", 1, TimeUnit.DAYS)) {
            // 处理消息
            processMessage(message);
        } else {
            // 重复消息，忽略
            System.out.println("重复消息: " + messageId);
        }
    }
}

五、集群搭建

5.1 普通集群

1
2
3
4
5
6
7
8
# 节点1
rabbitmq-server -detached

# 节点2
rabbitmq-server -detached
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

5.2 镜像队列

1
2
3
4
# 设置镜像策略
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all","ha-sync-mode":"automatic"}'

# 或通过Web管理界面设置

六、性能优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
1. 批量发送
List<String> messages = ...;
for (String msg : messages) {
    channel.basicPublish("", "queue", null, msg.getBytes());
}

2. 预取数量
channel.basicQos(10); // 一次预取10条

3. 连接池
// 使用连接池管理连接

4. 持久化策略
// 非关键消息不持久化

5. 懒加载队列
Map<String, Object> args = new HashMap<>();
args.put("x-queue-mode", "lazy");
channel.queueDeclare("lazy_queue", true, false, false, args);

七、监控告警

1
2
3
4
5
6
7
8
9
10
11
# 查看队列状态
rabbitmqctl list_queues

# 查看连接
rabbitmqctl list_connections

# 查看消费者
rabbitmqctl list_consumers

# Web管理界面
http://localhost:15672

总结

RabbitMQ高级特性：

  1. 死信队列：处理异常消息
  2. 延迟队列：实现定时任务
  3. 优先级队列：紧急消息优先处理
  4. 消息幂等：防止重复消费
  5. 集群搭建：高可用部署

参考资料