在nodejs中使用RabbitMQ（三）Routing、Topics、Headers

示例一、Routing

exchange类型direct，根据消息的routekey将消息直接转发到指定队列。producer.ts 生产者主要发送消息，consumer.ts负责接收消息，同时也都可以创建exchange交换机，创建队列，为队列绑定exchange，为避免重复简化代码，提高可维护性，队列相关操作移动到消费者端。队列，exchange交换机推荐在启动程序前手动创建好。

producer.ts

javascript">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 {let args = ['info', 'warn', 'error'];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)=>{}// ));const routeKey = args[Math.floor(Math.random() * 3)];console.log('消息发送是否成功', channel.publish(exchangeName,routeKey,Buffer.from(`发送消息,${i}${Math.ceil(Math.random() * 100000)},${routeKey}`),{ persistent: true },));}}});});setTimeout(() => {conn.close();process.exit(0);}, 1000);});
}start();

consumer.ts

javascript">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) => {if (err2) {console.log('队列创建失败', err2);return;}console.log('[*] waiting...');// 一次只有一个未确认消息，防止消费者过载channel.prefetch(1);channel.bindQueue(queueName, 'exchange1', 'info', {}, (err3, ok) => {console.log(queueName, '队列绑定结果', err3, ok);});channel.bindQueue(queueName, 'exchange1', 'warn', {}, (err3, ok) => {console.log(queueName, '队列绑定结果', err3, ok);});channel.bindQueue(queueName, 'exchange1', 'error', {}, (err3, ok) => {console.log(queueName, '队列绑定结果', err3, ok);});channel.consume(queueName,function (msg) {console.log('接收到的消息', msg?.content.toString());/*// 手动确认取消channel.ack(msg); noAck:false,// 自动确认消息// if (msg) {//   channel.ack(msg);// } */},{noAck: true, // 是否自动确认消息// noAck: false},(err3: any, ok: Replies.Empty) => {console.log(err3, ok);},);});});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

javascript">import RabbitMQ from 'amqplib';const conn = await RabbitMQ.connect('amqp://admin:admin1234@localhost:5672');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");
});const channel = await conn.createChannel();const queueName = 'queue2';await channel.assertQueue(queueName, { durable: true });console.log('[*] waiting...');// 一次只有一个未确认消息，防止消费者过载
await channel.prefetch(1);await channel.bindQueue(queueName, 'exchange1', 'error', {});channel.consume(queueName,function (msg) {console.log('接收到的消息', msg?.content.toString());/*// 手动确认取消channel.ack(msg); noAck:false,// 自动确认消息// if (msg) {//   channel.ack(msg);// } */},{noAck: true, // 是否自动确认消息// noAck: false},
);

示例二、Topic

exchange的topic类型和direct类似，使用的仍然是routeKey进行匹配转发，topic支持通过*和#进行模糊查询。*代码一个具体单词,#代码0或多个单词。

producer.ts

javascript">import RabbitMQ from 'amqplib';async function start() {const conn = await RabbitMQ.connect("amqp://admin:admin1234@localhost:5672?heartbeat=60");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);});try {const channel = await conn.createChannel();console.log("[AMQP] connected");const exchangeName = 'exchange4';await channel.assertExchange(exchangeName, 'topic', { durable: true });let args = ['123.orange.456', '123.456.rabbit', 'lazy', 'lazy.123', 'lazy.123.456'];for (let i = 0; i < 20; ++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)=>{}// ));const routeKey = args[Math.floor(Math.random() * args.length)];console.log('消息发送是否成功', channel.publish(exchangeName,routeKey,Buffer.from(`发送消息,${i}${Math.ceil(Math.random() * 100000)},${routeKey}`),{ persistent: true },));}} catch (err) {console.error("[AMQP]", err);return setTimeout(start, 1000);}setTimeout(() => {conn.close();process.exit(0);}, 1000);
}start();

consumer.ts

javascript">import RabbitMQ from 'amqplib';const conn = await RabbitMQ.connect('amqp://admin:admin1234@localhost:5672');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");
});const channel = await conn.createChannel();const queueName = 'queue1';channel.assertQueue(queueName, { durable: true });console.log('[*] waiting...');// 一次只有一个未确认消息，防止消费者过载
await channel.prefetch(1);// *代码一个具体单词,#代码0或多个单词
await channel.bindQueue(queueName, 'exchange4', '*.orange.*', {});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
});

consumer2.ts

javascript">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 });console.log('[*] waiting...');// 一次只有一个未确认消息，防止消费者过载channel.prefetch(1);channel.bindQueue(queueName, 'exchange4', '*.*.rabbit', {}, (err, ok) => {console.log(queueName, '队列绑定结果', err, ok);});channel.bindQueue(queueName, 'exchange4', 'lazy.#', {}, (err, ok) => {console.log(queueName, '队列绑定结果', err, 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});// return，error事件不会把消息重新放回队列channel.on('return', (msg) => {console.error('消息发送失败:', msg);});channel.on('error', (err) => {console.error('通道发生错误:', err);});});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");});
});

示例三、Headers

exchange类型headers，根据传递的头部信息进行转发，头部信息类型为object对象。在头部信息中要设置x-match属性，'x-match': 'any'， any，下方消息匹配上一个就可以。all，下方消息要全部匹配。

producer.ts

javascript">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 = 'exchange5';channel.assertExchange(exchangeName,'headers',{durable: true},(err, ok) => {if (err) {console.log('exchange路由转发创建失败', err);} else {let args = [{// 'x-match': 'any', // any,下方消息匹配上一个就可以; all,下方消息要全部匹配'loglevel': 'info',// 'buslevel': 'product',// 'syslevel': 'admin'},{// 'x-match': 'any', // any,下方消息匹配上一个就可以; all,下方消息要全部匹配// 'loglevel': 'info','buslevel': 'product','syslevel': 'admin'},{// 'x-match': 'any', // any,下方消息匹配上一个就可以; all,下方消息要全部匹配// 'loglevel': 'info','buslevel': 'product',// 'syslevel': 'admin'},{// 'x-match': 'all', // any,下方消息匹配上一个就可以; all,下方消息要全部匹配'loglevel': 'info','buslevel': 'product','syslevel': 'admin'},];for (let i = 0; i < 20; ++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)=>{}// ));const routeKey = args[Math.floor(Math.random() * args.length)];console.log('消息发送是否成功', routeKey, channel.publish(exchangeName,'',Buffer.from(`发送消息,${i}${Math.ceil(Math.random() * 100000)},${JSON.stringify(routeKey)}`),{persistent: true,headers: routeKey},));}}});});setTimeout(() => {conn.close();process.exit(0);}, 1000);});
}start();

consumer.ts

javascript">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 });console.log('[*] waiting...');// 一次只有一个未确认消息，防止消费者过载channel.prefetch(1);// *代码一个具体单词,#代码0或多个单词channel.bindQueue(queueName,'exchange5','',{'x-match': 'any', // any,下方消息匹配上一个就可以; all,下方消息要全部匹配'loglevel': 'info','buslevel': 'product','syslevel': 'admin'},(err, ok) => {console.log(queueName, '队列绑定结果', err, 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}, (err: any, ok: Replies.Empty) => {console.log(err, ok);});// return，error事件不会把消息重新放回队列channel.on('return', (msg) => {console.error('消息发送失败:', msg);});channel.on('error', (err) => {console.error('通道发生错误:', err);});});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

javascript">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 });console.log('[*] waiting...');// 一次只有一个未确认消息，防止消费者过载channel.prefetch(1);channel.bindQueue(queueName,'exchange5','',{'x-match': 'all', // any,下方消息匹配上一个就可以; all,下方消息要全部匹配'loglevel': 'info','buslevel': 'product','syslevel': 'admin'},(err, ok) => {console.log(queueName, '队列绑定结果', err, ok);},);channel.consume(queueName,function (msg) {console.log('接收到的消息', msg?.content.toString());/*// 手动确认取消channel.ack(msg); noAck:false,// 自动确认消息// if (msg) {//   channel.ack(msg);// } */},{noAck: true, // 是否自动确认消息// noAck: false},(err: any, ok: Replies.Empty) => {console.log(err, ok);},);// return，error事件不会把消息重新放回队列channel.on('return', (msg) => {console.error('消息发送失败:', msg);});channel.on('error', (err) => {console.error('通道发生错误:', err);});});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");});
});