多重绑定与路由

多重绑定允许队列从多个路由规则接收消息，实现消息聚合与多源数据整合。

定义

多重绑定指一个队列绑定到多个交换机，或绑定到同一交换机的多个路由键。队列会接收所有匹配路由的消息，消费者需能处理多种消息类型。

绑定模式

1. 单交换机多路由键

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public class MultiRoutingKeyExample {
    private static final String EXCHANGE = "order_exchange";
    private static final String QUEUE = "order_aggregator_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            channel.exchangeDeclare(EXCHANGE, "direct", true);
            channel.queueDeclare(QUEUE, true, false, false, null);
            
            // 一个队列绑定多个路由键
            channel.queueBind(QUEUE, EXCHANGE, "order.created");
            channel.queueBind(QUEUE, EXCHANGE, "order.updated");
            channel.queueBind(QUEUE, EXCHANGE, "order.cancelled");
            
            System.out.println("Queue bound to 3 routing keys");
            
            // 消费者接收所有路由键的消息
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String routingKey = delivery.getEnvelope().getRoutingKey();
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println("[" + routingKey + "] " + message);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            };
            
            channel.basicConsume(QUEUE, false, deliverCallback, consumerTag -> {});
        }
    }
}

2. 多交换机绑定

public class MultiExchangeExample {
    private static final String QUEUE = "shared_audit_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            // 声明多个交换机
            channel.exchangeDeclare("order_exchange", "direct", true);
            channel.exchangeDeclare("payment_exchange", "direct", true);
            channel.exchangeDeclare("user_exchange", "fanout", true);
            
            channel.queueDeclare(QUEUE, true, false, false, null);
            
            // 队列绑定到多个交换机
            channel.queueBind(QUEUE, "order_exchange", "order.*");
            channel.queueBind(QUEUE, "payment_exchange", "payment.completed");
            channel.queueBind(QUEUE, "user_exchange", "");
            
            System.out.println("Queue bound to 3 exchanges");
        }
    }
}

3. Topic通配符多重匹配

public class TopicMultiBindingExample {
    private static final String EXCHANGE = "log_topic";
    private static final String QUEUE = "comprehensive_log_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            channel.exchangeDeclare(EXCHANGE, "topic", true);
            channel.queueDeclare(QUEUE, true, false, false, null);
            
            // 使用多个通配符绑定
            channel.queueBind(QUEUE, EXCHANGE, "*.error");       // 所有错误日志
            channel.queueBind(QUEUE, EXCHANGE, "payment.#");     // 所有支付消息
            channel.queueBind(QUEUE, EXCHANGE, "order.us.*");    // 美国订单
            
            System.out.println("Topic multi-binding configured");
        }
    }
}

消息去重处理

多重绑定可能导致同一消息被多次投递到队列（当多个绑定规则匹配同一消息时）：

import com.rabbitmq.client.*;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class DeduplicationConsumer {
    private static final String QUEUE = "dedup_queue";
    private static final Set<String> processedIds = ConcurrentHashMap.newKeySet();

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            String messageId = delivery.getProperties().getMessageId();
            
            // 简单去重逻辑
            if (messageId != null && !processedIds.add(messageId)) {
                System.out.println("Duplicate message skipped: " + messageId);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                return;
            }
            
            System.out.println("Processing: " + message);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        
        channel.basicConsume(QUEUE, false, deliverCallback, consumerTag -> {});
    }
}

注意事项

• 多重绑定时，若多个规则匹配同一消息，队列会收到多条消息副本
• 消费者应能处理来自不同路由键或交换机的消息，做好类型判断
• 消息去重可通过messageId或业务唯一标识实现，避免重复处理
• 多重绑定增加路由复杂度，应控制绑定总数避免性能下降
• 解绑时需指定对应的交换机和路由键，逐一解除不需要的绑定

要点总结

• 多重绑定支持队列从多个路由规则接收消息，实现消息聚合
• 可绑定到同一交换机的多个路由键，或多个不同交换机
• Topic通配符多重绑定需警惕同一消息被多次投递
• 消费者应实现消息去重和类型判断逻辑
• 控制绑定总数，避免路由匹配性能下降

文章存放路径：D:\git2\jwdev\articles\RABBITMQ\进阶\消息路由与绑定\多重绑定与路由.md