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 示例

1. const amqp = require('amqplib/callback_api');
3. amqp.connect('amqp://localhost', function(error0, connection) {
4.   if (error0) {
5.     throw error0;
6.   }
7.   connection.createChannel(function(error1, channel) {
8.     if (error1) {
9.       throw error1;
10.     }
11.     const exchange = 'direct_logs';
12.     const msg = 'Hello World!';
13.     const routingKey = 'key1';
15.     channel.assertExchange(exchange, 'direct', { durable: true });
16.     channel.publish(exchange, routingKey, Buffer.from(msg));
17.     console.log(" [x] Sent %s: '%s'", routingKey, msg);
18.   });
20.   setTimeout(function() {
21.     connection.close();
22.     process.exit(0);
23.   }, 500);
24. });

复制代码

Fanout Exchange 示例

1. const amqp = require('amqplib/callback_api');
3. amqp.connect('amqp://localhost', function(error0, connection) {
4.   if (error0) {
5.     throw error0;
6.   }
7.   connection.createChannel(function(error1, channel) {
8.     if (error1) {
9.       throw error1;
10.     }
11.     const exchange = 'logs';
12.     const msg = 'Hello World!';
14.     channel.assertExchange(exchange, 'fanout', { durable: true });
15.     channel.publish(exchange, '', Buffer.from(msg));
16.     console.log(" [x] Sent %s", msg);
17.   });
19.   setTimeout(function() {
20.     connection.close();
21.     process.exit(0);
22.   }, 500);
23. });

复制代码

Topic Exchange 示例

Topic Exchange 允许使用通配符进行路由，支持更复杂的路由规则。
发布者代码

1. const amqp = require('amqplib/callback_api');
3. amqp.connect('amqp://localhost', function(error0, connection) {
4.   if (error0) {
5.     throw error0;
6.   }
7.   connection.createChannel(function(error1, channel) {
8.     if (error1) {
9.       throw error1;
10.     }
11.     const exchange = 'topic_logs';
12.     const msg = 'Hello World!';
13.     const routingKey = 'quick.orange.rabbit';
15.     channel.assertExchange(exchange, 'topic', { durable: true });
16.     channel.publish(exchange, routingKey, Buffer.from(msg));
17.     console.log(" [x] Sent %s: '%s'", routingKey, msg);
18.   });
20.   setTimeout(function() {
21.     connection.close();
22.     process.exit(0);
23.   }, 500);
24. });

复制代码

消费者代码

1. const amqp = require('amqplib/callback_api');
3. amqp.connect('amqp://localhost', function(error0, connection) {
4.   if (error0) {
5.     throw error0;
6.   }
7.   connection.createChannel(function(error1, channel) {
8.     if (error1) {
9.       throw error1;
10.     }
11.     const exchange = 'topic_logs';
12.     const queue = 'topic_queue';
14.     channel.assertExchange(exchange, 'topic', { durable: true });
15.     channel.assertQueue(queue, { durable: true });
17.     // 绑定队列到交换机，使用通配符
18.     channel.bindQueue(queue, exchange, '*.orange.*');
20.     channel.consume(queue, function(msg) {
21.       if (msg.content) {
22.         console.log(" [x] Received %s: '%s'", msg.fields.routingKey, msg.content.toString());
23.       }
24.     }, { noAck: true });
25.   });
26. });

复制代码

在这个示例中，发布者将消息发送到 topic_logs 互换机，使用路由键 quick.orange.rabbit。消费者绑定到 topic_logs 互换机，使用通配符 *.orange.*，因此会接收到全部包含 orange 的消息。
Headers Exchange 示例

Headers Exchange 基于消息头部属性进行路由，适用于需要复杂路由规则的场景。
发布者代码

1. const amqp = require('amqplib/callback_api');
3. amqp.connect('amqp://localhost', function(error0, connection) {
4.   if (error0) {
5.     throw error0;
6.   }
7.   connection.createChannel(function(error1, channel) {
8.     if (error1) {
9.       throw error1;
10.     }
11.     const exchange = 'headers_logs';
12.     const msg = 'Hello World!';
14.     channel.assertExchange(exchange, 'headers', { durable: true });
15.     channel.publish(exchange, '', Buffer.from(msg), {
16.       headers: {
17.         'format': 'pdf',
18.         'type': 'report'
19.       }
20.     });
21.     console.log(" [x] Sent %s", msg);
22.   });
24.   setTimeout(function() {
25.     connection.close();
26.     process.exit(0);
27.   }, 500);
28. });

复制代码

消费者代码

1. const amqp = require('amqplib/callback_api');
3. amqp.connect('amqp://localhost', function(error0, connection) {
4.   if (error0) {
5.     throw error0;
6.   }
7.   connection.createChannel(function(error1, channel) {
8.     if (error1) {
9.       throw error1;
10.     }
11.     const exchange = 'headers_logs';
12.     const queue = 'headers_queue';
14.     channel.assertExchange(exchange, 'headers', { durable: true });
15.     channel.assertQueue(queue, { durable: true });
17.     // 绑定队列到交换机，使用头部属性
18.     channel.bindQueue(queue, exchange, '', {
19.       'x-match': 'all',
20.       'format': 'pdf',
21.       'type': 'report'
22.     });
24.     channel.consume(queue, function(msg) {
25.       if (msg.content) {
26.         console.log(" [x] Received %s", msg.content.toString());
27.       }
28.     }, { noAck: true });
29.   });
30. });

复制代码

在这个示例中，发布者将消息发送到 headers_logs 互换机，并设置消息头部属性 format: pdf 和 type: report。消费者绑定到 headers_logs 互换机，使用头部属性匹配 format: pdf 和 type: report，因此会接收到符合这些头部属性的消息。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。