本文目录导读：
在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 -> {});
注意事项
- 绑定顺序：先声明交换机和队列，再进行绑定
- 持久化：交换机、队列、消息都可以设置持久化
- 重复绑定：相同的绑定不会产生错误，但可能会有性能影响
- 路由键：不同交换机类型对路由键的处理不同
- 资源清理：使用后记得关闭连接和通道
这些示例涵盖了Java中使用RabbitMQ进行交换机-队列绑定的主要场景，可以根据实际需求选择合适的交换机类型和绑定方式。
