医院信息化建设网站梵克雅宝手链

exhcange交换机类型fanout和direct对比总结

特性	Fanout 交换机	Direct 交换机（默认类型）
消息路由方式	广播，忽略路由键	精确匹配路由键
绑定关系	所有绑定到该交换机的队列都接收消息	只有路由键匹配的队列接收消息
使用场景	日志系统、事件通知	任务分配、错误处理
示例	所有服务都需要接收相同的日志或通知	不同类型的日志或任务被路由到不同队列

示例一 

使用Fanout 交换机实现广播，它不考虑任何路由键（routing key），而是将接收到的消息广播到所有绑定到该交换机的队列中。

producer.ts 发送消息，exchange类型fanout，队列和消息没有设置持久化，如果有消费者会接收到数据，没有消费者数据会丢失，重启rabbitmq数据会丢失。

import amqp from 'amqplib/callback_api';amqp.connect('amqp://admin:admin1234@localhost:5672', function (error0, connection) {if (error0) {throw error0;}connection.createChannel(function (error1, channel) {if (error1) {throw error1;}var exchange = 'exchange3';var msg = 'Hello World!';channel.assertExchange(exchange, 'fanout', {durable: false});channel.publish(exchange, '', Buffer.from(msg));console.log(" [x] Sent %s", msg);});setTimeout(function () {connection.close();process.exit(0);}, 500);
});

consumer.ts

import amqp from 'amqplib/callback_api';amqp.connect('amqp://admin:admin1234@localhost:5672', function (error0, connection) {if (error0) {throw error0;}connection.createChannel(function (error1, channel) {if (error1) {throw error1;}var exchange = 'exchange3';channel.assertExchange(exchange, 'fanout', {durable: false});channel.assertQueue('', {exclusive: true}, function (error2, q) {if (error2) {throw error2;}console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q.queue);channel.bindQueue(q.queue, exchange, '');channel.consume(q.queue, function (msg) {if (msg?.content) {console.log(" [x] %s", msg.content.toString());}}, {noAck: true});});});
});

 示例二

使用direct类型交换机，只有路由键匹配的队列会接收到消息（精确匹配），适合需要根据特定条件（如路由键）将消息路由到特定队列的场景，如任务分配、错误处理等。

producer.ts 发送消息routeKey指定为了route.key，设置了队列和消息持久化，publish发送时指定了route.key。

import RabbitMQ from 'amqplib/callback_api';function start() {RabbitMQ.connect("amqp://admin:admin1234@localhost:5672?heartbeat=60", function (err0, conn) {if (err0) {console.error("[AMQP]", err0.message);return setTimeout(start, 1000);}conn.on("error", function (err1) {if (err1.message !== "Connection closing") {console.error("[AMQP] conn error", err1.message);}});conn.on("close", function () {console.error("[AMQP] reconnecting");return setTimeout(start, 1000);});console.log("[AMQP] connected");conn.createChannel(async (err2, channel) => {if (err2) {console.error("[AMQP]", err2.message);return setTimeout(start, 1000);}const exchangeName = 'exchange1';channel.assertExchange(exchangeName,'direct',{durable: true},(err, ok) => {if (err) {console.log('exchange路由转发创建失败', err);} else {console.log('exchange路由转发创建成功', ok);for (let i = 0; i < 10; ++i) {// 给单个队列发送消息// console.log('message send!', channel.sendToQueue(//   queueName,//   Buffer.from(`发送消息,${i}${Math.ceil(Math.random() * 100000)}`),//   { persistent: true, correlationId: 'ooooooooooooooo' },// 消息持久化，重启后存在。//   // (err: any, ok: Replies.Empty)=>{}// ));console.log('消息发送是否成功', channel.publish(exchangeName,'route.key',Buffer.from(`发送消息,${i}${Math.ceil(Math.random() * 100000)}`),{ persistent: true },));}}});});setTimeout(() => {conn.close();process.exit(0);}, 1000);});
}start();

consumer.ts 绑定了exchange并指定了route.key，可以接收到生产者发送的消息。

import RabbitMQ, { type Replies } from 'amqplib/callback_api';RabbitMQ.connect('amqp://admin:admin1234@localhost:5672', (err0, conn) => {if (err0) {console.error(err0);return;}conn.createChannel(function (err1, channel) {const queueName = 'queue1';channel.assertQueue(queueName, { durable: true }, (err2, ok) => {if (err2) {console.log('队列创建失败', err2);return;}console.log('[*] waiting...');// 一次只有一个未确认消息，防止消费者过载channel.prefetch(1);channel.bindQueue(queueName, 'exchange1', 'route.key', {}, (err3, ok) => {console.log(queueName, '队列绑定结果', err3, ok);});channel.consume(queueName, function (msg) {console.log('接收到的消息', msg?.content.toString());// 手动确认取消channel.ack(msg);设置noAck:false,// 自动确认消息noAck:true,不需要channel.ack(msg);try {if (msg) {channel.ack(msg);}} catch (err) {if (msg) {// 第二个参数，false拒绝当前消息// 第二个参数，true拒绝小于等于当前消息// 第三个参数，3false从队列中清除// 第三个参数，4true从新在队列中排队channel.nack(msg, false, false);}console.log(err);}}, {// noAck: true, // 是否自动确认消息，为true不需要调用channel.ack(msg);noAck: false});});});conn.on("error", function (err1) {if (err1.message !== "Connection closing") {console.error("[AMQP] conn error", err1.message);}});conn.on("close", function () {console.error("[AMQP] reconnecting");});
});

 consumer2.ts

import RabbitMQ, { type Replies } from 'amqplib/callback_api';RabbitMQ.connect('amqp://admin:admin1234@localhost:5672', (err0, conn) => {if (err0) {console.error(err0);return;}conn.createChannel(function (err1, channel) {const queueName = 'queue2';channel.assertQueue(queueName, { durable: true }, (err2, ok) => {if (err2) {console.log('队列创建失败', err2);return;}console.log('[*] waiting...');// 一次只有一个未确认消息，防止消费者过载channel.prefetch(1);channel.bindQueue(ok.queue, 'exchange1', 'route.key', {}, (err3, ok) => {console.log(queueName, '队列绑定结果', err3, ok);});channel.consume(ok.queue, function (msg) {console.log('接收到的消息', msg?.content.toString());// 手动确认取消channel.ack(msg);设置noAck:false,// 自动确认消息noAck:true,不需要channel.ack(msg);try {if (msg) {channel.ack(msg);}} catch (err) {if (msg) {// 第二个参数，false拒绝当前消息// 第二个参数，true拒绝小于等于当前消息// 第三个参数，3false从队列中清除// 第三个参数，4true从新在队列中排队channel.nack(msg, false, false);}console.log(err);}}, {// noAck: true, // 是否自动确认消息，为true不需要调用channel.ack(msg);noAck: false});});});conn.on("error", function (err1) {if (err1.message !== "Connection closing") {console.error("[AMQP] conn error", err1.message);}});conn.on("close", function () {console.error("[AMQP] reconnecting");});
});

consumer3.ts

import RabbitMQ, { type Replies } from 'amqplib/callback_api';RabbitMQ.connect('amqp://admin:admin1234@localhost:5672', (err0, conn) => {if (err0) {console.error(err0);return;}conn.createChannel(function (err1, channel) {const queueName = 'queue3';channel.assertQueue(queueName, { durable: true }, (err2, ok) => {if (err2) {console.log('队列创建失败', err2);return;}console.log('[*] waiting...');// 一次只有一个未确认消息，防止消费者过载channel.prefetch(1);// 绑定的routeKey不同不会接收到route.key发送的消息channel.bindQueue(queueName, 'exchange1', 'routeKey', {}, (err3, ok) => {console.log(queueName, '队列绑定结果', err3, ok);});channel.consume(queueName, function (msg) {console.log('接收到的消息', msg?.content.toString());// 手动确认取消channel.ack(msg);设置noAck:false,// 自动确认消息noAck:true,不需要channel.ack(msg);try {if (msg) {channel.ack(msg);}} catch (err) {if (msg) {// 第二个参数，false拒绝当前消息// 第二个参数，true拒绝小于等于当前消息// 第三个参数，3false从队列中清除// 第三个参数，4true从新在队列中排队channel.nack(msg, false, false);}console.log(err);}}, {// noAck: true, // 是否自动确认消息，为true不需要调用channel.ack(msg);noAck: false});});});conn.on("error", function (err1) {if (err1.message !== "Connection closing") {console.error("[AMQP] conn error", err1.message);}});conn.on("close", function () {console.error("[AMQP] reconnecting");});
});