Java交换机案例如何绑定队列

wen java案例 2026-06-30 218

本文目录导读：

Java交换机案例如何绑定队列

基本绑定示例
不同交换机类型的绑定示例
完整的生产者-消费者示例
其他绑定方法
注意事项

在Java中绑定交换机和队列通常使用RabbitMQ客户端,以下是完整的绑定示例：

基本绑定示例

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class RabbitMQBindingExample {
    public static void main(String[] args) throws Exception {
        // 1. 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 2. 创建连接和通道
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 3. 声明交换机（如果不存在则创建）
            String exchangeName = "my_exchange";
            String exchangeType = "direct"; // direct, topic, fanout, headers
            channel.exchangeDeclare(exchangeName, exchangeType, true);
            // 4. 声明队列（如果不存在则创建）
            String queueName = "my_queue";
            channel.queueDeclare(queueName, true, false, false, null);
            // 5. 绑定队列到交换机
            String routingKey = "my_routing_key";
            channel.queueBind(queueName, exchangeName, routingKey);
            System.out.println("队列 " + queueName + " 已绑定到交换机 " + exchangeName + 
                             "，路由键: " + routingKey);
        }
    }
}

不同交换机类型的绑定示例

Direct Exchange（直接交换机）

// 交换机声明
channel.exchangeDeclare("direct_exchange", "direct", true);
// 队列声明
channel.queueDeclare("order_queue", true, false, false, null);
channel.queueDeclare("payment_queue", true, false, false, null);
// 绑定队列到指定路由键
channel.queueBind("order_queue", "direct_exchange", "order.create");
channel.queueBind("payment_queue", "direct_exchange", "payment.success");
// 发送消息时指定路由键
channel.basicPublish("direct_exchange", "order.create", null, message.getBytes());

Topic Exchange（主题交换机）

// 交换机声明
channel.exchangeDeclare("topic_exchange", "topic", true);
// 队列声明
channel.queueDeclare("all_logs_queue", true, false, false, null);
channel.queueDeclare("error_logs_queue", true, false, false, null);
// 使用通配符绑定
channel.queueBind("all_logs_queue", "topic_exchange", "log.*");      // 匹配所有日志
channel.queueBind("error_logs_queue", "topic_exchange", "log.error"); // 只匹配错误日志
// 通配符规则：
// * 匹配一个单词
// # 匹配零个或多个单词
// log.*       匹配 log.info, log.error, log.warn
// log.#       匹配 log, log.info, log.error.database

Fanout Exchange（扇形交换机）

// 交换机声明 - 广播到所有绑定的队列
channel.exchangeDeclare("fanout_exchange", "fanout", true);
// 队列声明
channel.queueDeclare("queue1", true, false, false, null);
channel.queueDeclare("queue2", true, false, false, null);
// 绑定 - 路由键被忽略（传空字符串或null）
channel.queueBind("queue1", "fanout_exchange", "");
channel.queueBind("queue2", "fanout_exchange", "");
// 发送消息 - 所有绑定的队列都会收到
channel.basicPublish("fanout_exchange", "", null, "广播消息".getBytes());

Headers Exchange（头交换机）

import java.util.HashMap;
import java.util.Map;
// 交换机声明
channel.exchangeDeclare("headers_exchange", "headers", true);
// 队列声明
channel.queueDeclare("queue_headers", true, false, false, null);
// 设置绑定参数
Map<String, Object> headers = new HashMap<>();
headers.put("type", "order");
headers.put("priority", "high");
// 使用头匹配绑定
channel.queueBind("queue_headers", "headers_exchange", "", 
                  new HashMap<String, Object>() {{
                      put("x-match", "all"); // all = 所有头都匹配, any = 任意一个匹配
                      put("type", "order");
                      put("priority", "high");
                  }});
// 发送消息时设置头
Map<String, Object> messageHeaders = new HashMap<>();
messageHeaders.put("type", "order");
messageHeaders.put("priority", "high");
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .headers(messageHeaders)
    .build();
channel.basicPublish("headers_exchange", "", props, "头匹配消息".getBytes());

完整的生产者-消费者示例

生产者代码

public class MessageProducer {
    private final static String EXCHANGE_NAME = "demo_exchange";
    private final static String QUEUE_NAME = "demo_queue";
    private final static String ROUTING_KEY = "demo.key";
    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_NAME, "direct", true);
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            // 绑定
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
            // 发送消息
            String message = "Hello RabbitMQ!";
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, message.getBytes());
            System.out.println(" [x] 发送 '" + message + "'");
        }
    }
}

消费者代码

public class MessageConsumer {
    private final static String EXCHANGE_NAME = "demo_exchange";
    private final static String QUEUE_NAME = "demo_queue";
    private final static String ROUTING_KEY = "demo.key";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 声明交换机（需与生产者一致）
        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
        // 声明队列（需与生产者一致或获取服务端生成的队列名）
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        // 绑定
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
        // 消费消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] 收到 '" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
        System.out.println(" [*] 等待消息...");
    }
}

其他绑定方法

批量绑定

// 单个队列绑定多个路由键
channel.queueBind("queue", "exchange", "key1");
channel.queueBind("queue", "exchange", "key2");
channel.queueBind("queue", "exchange", "key3");
// 或者使用循环
String[] routingKeys = {"create", "update", "delete"};
for (String key : routingKeys) {
    channel.queueBind("queue", "exchange", key);
}

解绑操作

// 解绑队列和交换机的绑定
channel.queueUnbind("queue", "exchange", "routing_key");
// 删除交换机
channel.exchangeDelete("exchange_name");
// 删除队列
channel.queueDelete("queue_name");

使用临时队列

// 声明临时队列（系统自动生成名称）
String tempQueue = channel.queueDeclare().getQueue();
// 绑定临时队列到交换机
channel.queueBind(tempQueue, "exchange", "routing_key");
// 消费者使用临时队列
channel.basicConsume(tempQueue, true, deliverCallback, consumerTag -> {});