在 RabbitMQ 中，交換機（Exchange）是一個核心組件，負責接收來自生產(chǎn)者的消息，并根據(jù)特定的路由規(guī)則將消息分發(fā)到相應的隊列。交換機的存在改變了消息發(fā)送的模式，使得消息的路由更加靈活和高效。

交換機的類型

RabbitMQ 提供了四種主要類型的交換機，每種交換機的路由規(guī)則不同：

1. Direct Exchange（直連交換機）：

   • 功能：基于路由鍵（Routing Key）將消息發(fā)送到與該路由鍵完全匹配的隊列。
   • 應用場景：適用于需要精確匹配路由鍵的場景。
   • 示例：假設有兩個隊列 A 和 B，A 綁定了路由鍵 key1，B 綁定了路由鍵 key2。當生產(chǎn)者發(fā)送一條路由鍵為 key1 的消息時，只有隊列 A 會接收到這條消息。

2. Fanout Exchange（扇出交換機）：

   • 功能：將消息廣播到所有綁定到該交換機的隊列，不考慮路由鍵。
   • 應用場景：適用于需要將消息廣播到多個隊列的場景。
   • 示例：假設有兩個隊列 A 和 B 都綁定到了一個 Fanout 交換機上。當生產(chǎn)者發(fā)送一條消息到該交換機時，A 和 B 都會接收到這條消息。

3. Topic Exchange（主題交換機）：

   • 功能：基于路由鍵的模式匹配（使用通配符）將消息發(fā)送到匹配的隊列。
   • 應用場景：適用于需要基于模式匹配路由鍵的場景。
   • 示例：假設有兩個隊列 A 和 B，A 綁定了路由鍵模式 key.*，B 綁定了路由鍵模式 key.#。當生產(chǎn)者發(fā)送一條路由鍵為 key.test 的消息時，A 和 B 都會接收到這條消息。

4. Headers Exchange（頭交換機）：

   • 功能：基于消息的頭部屬性進行匹配，將消息發(fā)送到匹配的隊列。
   • 應用場景：適用于需要基于消息頭部屬性進行路由的場景。
   • 示例：這種交換機使用較少，通常在特定情況下才會使用。

交換機的作用

• 消息路由：交換機根據(jù)路由規(guī)則將消息分發(fā)到相應的隊列。
• 解耦生產(chǎn)者和消費者：生產(chǎn)者只需將消息發(fā)送到交換機，不需要知道消息的最終目的地隊列。
• 靈活性和擴展性：通過不同類型的交換機，可以實現(xiàn)復雜的消息路由邏輯，滿足各種業(yè)務需求。

示例代碼

以下是如何使用 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 允許使用通配符進行路由，支持更復雜的路由規(guī)則。

發(fā)布者代碼

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 });});
});

在這個示例中，發(fā)布者將消息發(fā)送到 topic_logs 交換機，使用路由鍵 quick.orange.rabbit。消費者綁定到 topic_logs 交換機，使用通配符 *.orange.*，因此會接收到所有包含 orange 的消息。

Headers Exchange 示例

Headers Exchange 基于消息頭部屬性進行路由，適用于需要復雜路由規(guī)則的場景。

發(fā)布者代碼

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 });});
});

在這個示例中，發(fā)布者將消息發(fā)送到 headers_logs 交換機，并設置消息頭部屬性 format: pdf 和 type: report。消費者綁定到 headers_logs 交換機，使用頭部屬性匹配 format: pdf 和 type: report，因此會接收到符合這些頭部屬性的消息。