当前位置： 首页 > wzjs >正文

上海网站建设好处百度店铺

wzjs 2025/8/20 3:26:32
上海网站建设好处,百度店铺,网络诈骗做网站 维护,营销型网站建设工资示例一、Routing exchange类型direct，根据消息的routekey将消息直接转发到指定队列。producer.ts 生产者主要发送消息，consumer.ts负责接收消息，同时也都可以创建exchange交换机，创建队列，为队列绑定exchange&#xff…

示例一、Routing

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

producer.ts 

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

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

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

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

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

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

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

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

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");});
});
http://www.dtcms.com/wzjs/412833.html

相关文章：

  • 广州番禺网站建设公司百度地图的精准定位功能
  • 国网交流建设公司网站优化模型数学建模
  • 无锡高端网站建设app推广80元一单
  • 单位如何做网站宣传品牌网络推广
  • 网站开发时会遇到哪些问题北京seo做排名
  • 网站项目建设周期人工智能培训机构排名前十
  • 长春网站设计外包网络推广网站排行榜
  • 汽车网站建设价格武汉大学人民医院怎么样
  • 做购物网站收费市场调研报告
  • 用什么框架做网站快武汉网站优化公司
  • 宣传片剪辑优化营商环境心得体会
  • 网站降权怎么恢复今日国际新闻头条15条简短
  • 东莞优化网站建设福建seo外包
  • 自己做卖东西网站线上销售水果营销方案
  • 绵阳网站推广排名腾讯中国联通
  • 网站建设模板怎么做抖音seo怎么做的
  • 企业网站建设实训总结推广如何做网上引流
  • 网站建设维护学习宁波网络推广
  • 自己做网站怎么让字体居中网络营销就是
  • 网络工程专业学什么课程公司网站如何seo
  • 手机商城建站系统网站建设宁波正规seo快速排名公司
  • 哪个平台做网站好免费信息发布平台网站
  • 网站域名劫持如何提升网站搜索排名
  • 清理空壳网站东营网站建设费用
  • 网站常用的推广方法有哪些网站设计优化
  • 阿里巴巴国际网站怎么做网站建设
  • 如何修改网站后台登陆入口路劲飞猪关键词排名优化
  • 网络程序设计学什么北京优化seo排名
  • 河南网站制作公司打开2345网址大全
  • 网站制作与免费网站建设微信软文范例大全100