RabbitMQ 交换机的类型
在 RabbitMQ 中，交换机（Exchange）是一个核心组件，负责接收来自生产者的消息，并根据特定的路由规则将消息分发到相应的队列。交换机的存在改变了消息发送的模式，使得消息的路由更加灵活和高效。
交换机的类型
RabbitMQ 提供了四种主要类型的交换机，每种交换机的路由规则不同：
-
Direct Exchange（直连交换机）：
- 功能：基于路由键（Routing Key）将消息发送到与该路由键完全匹配的队列。
- 应用场景：适用于需要精确匹配路由键的场景。
- 示例：假设有两个队列 A 和 B，A 绑定了路由键
key1，B 绑定了路由键key2。当生产者发送一条路由键为key1的消息时，只有队列 A 会接收到这条消息。
-
Fanout Exchange（扇出交换机）：
- 功能：将消息广播到所有绑定到该交换机的队列，不考虑路由键。
- 应用场景：适用于需要将消息广播到多个队列的场景。
- 示例：假设有两个队列 A 和 B 都绑定到了一个 Fanout 交换机上。当生产者发送一条消息到该交换机时，A 和 B 都会接收到这条消息。
-
Topic Exchange（主题交换机）：
- 功能：基于路由键的模式匹配（使用通配符）将消息发送到匹配的队列。
- 应用场景：适用于需要基于模式匹配路由键的场景。
- 示例：假设有两个队列 A 和 B，A 绑定了路由键模式
key.*，B 绑定了路由键模式key.#。当生产者发送一条路由键为key.test的消息时，A 和 B 都会接收到这条消息。
-
Headers Exchange（头交换机）：
- 功能：基于消息的头部属性进行匹配，将消息发送到匹配的队列。
- 应用场景：适用于需要基于消息头部属性进行路由的场景。
- 示例：这种交换机使用较少，通常在特定情况下才会使用。
交换机的作用
- 消息路由：交换机根据路由规则将消息分发到相应的队列。
- 解耦生产者和消费者：生产者只需将消息发送到交换机，不需要知道消息的最终目的地队列。
- 灵活性和扩展性：通过不同类型的交换机，可以实现复杂的消息路由逻辑，满足各种业务需求。
示例代码
以下是如何使用 Direct Exchange 和 Fanout Exchange 的示例代码：
Direct Exchange 示例
const amqp = require('amqplib/callback_api');
amqp.connect('amqp://localhost', function(error0, connection) {
if (error0) {
throw error0;
}
connection.createChannel(function(error1, channel) {
if (error1) {
throw error1;
}
const exchange = 'direct_logs';
const msg = 'Hello World!';
const routingKey = 'key1';
channel.assertExchange(exchange, 'direct', { durable: true });
channel.publish(exchange, routingKey, Buffer.from(msg));
console.log(" [x] Sent %s: '%s'", routingKey, msg);
});
setTimeout(function() {
connection.close();
process.exit(0);
}, 500);
});
Fanout Exchange 示例
const amqp = require('amqplib/callback_api');
amqp.connect('amqp://localhost', function(error0, connection) {
if (error0) {
throw error0;
}
connection.createChannel(function(error1, channel) {
if (error1) {
throw error1;
}
const exchange = 'logs';
const msg = 'Hello World!';
channel.assertExchange(exchange, 'fanout', { durable: true });
channel.publish(exchange, '', Buffer.from(msg));
console.log(" [x] Sent %s", msg);
});
setTimeout(function() {
connection.close();
process.exit(0);
}, 500);
});
Topic Exchange 示例
Topic Exchange 允许使用通配符进行路由，支持更复杂的路由规则。
发布者代码
const amqp = require('amqplib/callback_api');
amqp.connect('amqp://localhost', function(error0, connection) {
if (error0) {
throw error0;
}
connection.createChannel(function(error1, channel) {
if (error1) {
throw error1;
}
const exchange = 'topic_logs';
const msg = 'Hello World!';
const routingKey = 'quick.orange.rabbit';
channel.assertExchange(exchange, 'topic', { durable: true });
channel.publish(exchange, routingKey, Buffer.from(msg));
console.log(" [x] Sent %s: '%s'", routingKey, msg);
});
setTimeout(function() {
connection.close();
process.exit(0);
}, 500);
});
消费者代码
const amqp = require('amqplib/callback_api');
amqp.connect('amqp://localhost', function(error0, connection) {
if (error0) {
throw error0;
}
connection.createChannel(function(error1, channel) {
if (error1) {
throw error1;
}
const exchange = 'topic_logs';
const queue = 'topic_queue';
channel.assertExchange(exchange, 'topic', { durable: true });
channel.assertQueue(queue, { durable: true });
// 绑定队列到交换机，使用通配符
channel.bindQueue(queue, exchange, '*.orange.*');
channel.consume(queue, function(msg) {
if (msg.content) {
console.log(" [x] Received %s: '%s'", msg.fields.routingKey, msg.content.toString());
}
}, { noAck: true });
});
});
在这个示例中，发布者将消息发送到 topic_logs 交换机，使用路由键 quick.orange.rabbit。消费者绑定到 topic_logs 交换机，使用通配符 *.orange.*，因此会接收到所有包含 orange 的消息。
Headers Exchange 示例
Headers Exchange 基于消息头部属性进行路由，适用于需要复杂路由规则的场景。
发布者代码
const amqp = require('amqplib/callback_api');
amqp.connect('amqp://localhost', function(error0, connection) {
if (error0) {
throw error0;
}
connection.createChannel(function(error1, channel) {
if (error1) {
throw error1;
}
const exchange = 'headers_logs';
const msg = 'Hello World!';
channel.assertExchange(exchange, 'headers', { durable: true });
channel.publish(exchange, '', Buffer.from(msg), {
headers: {
'format': 'pdf',
'type': 'report'
}
});
console.log(" [x] Sent %s", msg);
});
setTimeout(function() {
connection.close();
process.exit(0);
}, 500);
});
消费者代码
const amqp = require('amqplib/callback_api');
amqp.connect('amqp://localhost', function(error0, connection) {
if (error0) {
throw error0;
}
connection.createChannel(function(error1, channel) {
if (error1) {
throw error1;
}
const exchange = 'headers_logs';
const queue = 'headers_queue';
channel.assertExchange(exchange, 'headers', { durable: true });
channel.assertQueue(queue, { durable: true });
// 绑定队列到交换机，使用头部属性
channel.bindQueue(queue, exchange, '', {
'x-match': 'all',
'format': 'pdf',
'type': 'report'
});
channel.consume(queue, function(msg) {
if (msg.content) {
console.log(" [x] Received %s", msg.content.toString());
}
}, { noAck: true });
});
});
在这个示例中，发布者将消息发送到 headers_logs 交换机，并设置消息头部属性 format: pdf 和 type: report。消费者绑定到 headers_logs 交换机，使用头部属性匹配 format: pdf 和 type: report，因此会接收到符合这些头部属性的消息。