प्रौद्योगिकी साझेदारी

egg.js सन्देशपङ्क्ति rabbitMQ इत्यस्य उपयोगं करोति

2024-07-12

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

1. egg-amqplib: rabbitmq सन्देशपङ्क्तिसंक्षेपणम् आधारितं पुस्तकालयम्

प्रतिस्था:

npm i egg-amqplib --save

पवर्तयति

  1. // {app_root}/config/plugin.js
  2. exports.amqplib = {
  3.   enable: true,
  4.   package: 'egg-amqplib',
  5. };

स्थापयति

  1. // {app_root}/config/config.default.js
  2. exports.amqplib = {
  3.   client: {
  4.     // url: 'amqp://localhost',
  5.     connectOptions: {
  6.       protocol: 'amqp',
  7.       hostname: 'localhost',
  8.       port: 5672,
  9.       username: 'guest',
  10.       password: 'guest',
  11.       locale: 'en_US',
  12.       frameMax: 0,
  13.       heartbeat: 0,
  14.       vhost: '/',
  15.     },
  16.     // socketOptions: {
  17.     //   cert: certificateAsBuffer, // client cert
  18.     //   key: privateKeyAsBuffer, // client key
  19.     //   passphrase: 'MySecretPassword', // passphrase for key
  20.     //   ca: [caCertAsBuffer], // array of trusted CA certs
  21.     // },
  22.   },
  23. };

github पश्यन्तु:https://github.com/zubinzhang/अण्डा-अम्क्यूप्लिब

नियन्त्रणस्तरः : १.

  1. 'use strict';
  2.  
  3. const Controller = require('egg').Controller;
  4. const queueName = 'test';
  5.  
  6. class HomeController extends Controller {
  7.   async publish() {
  8.     const { msg } = this.ctx.query;
  9.  
  10.     const ch = await this.app.amqplib.createChannel();
  11.     await ch.assertQueue(queueName, { durable: false });
  12.     const ok = await ch.sendToQueue(queueName, Buffer.from(msg));
  13.     await ch.close();
  14.  
  15.     this.ctx.body = ok;
  16.     this.ctx.status = 200;
  17.   }
  18.  
  19.   async consume() {
  20.     const ch = await this.app.amqplib.createChannel();
  21.     await ch.assertQueue(queueName, { durable: false });
  22.     const msg = await new Promise(resolve => ch.consume(queueName, msg => resolve(msg)));
  23.  
  24.     if (msg !== null) {
  25.       ch.ack(msg);
  26.       await ch.close();
  27.  
  28.       this.ctx.status = 200;
  29.       this.ctx.body = { msg: msg.content.toString() };
  30.     } else {
  31.       this.ctx.status = 500;
  32.     }
  33.   }
  34. }
  35.  
  36. module.exports = HomeController;

मार्गनिर्धारणम् : १.

  1. 'use strict';
  2.  
  3. module.exports = app => {
  4.   const { router, controller } = app;
  5.  
  6.   router.get('/publish', controller.home.publish);
  7.   router.get('/consume', controller.home.consume);
  8. };

refer to : १.अण्डा-amqplib / परीक्षण / फिक्स्चर / ऐप्स / amqplib-परीक्षण / ऐप / नियंत्रक / home.js पर मास्टर · zubinzhang / अंडा-amqplib · GitHub

2. rabbitmq संस्थापयन्तु rabbitmq संस्थापयितुं docker इत्यस्य उपयोगं कर्तुं शक्नुवन्ति।

docker run --name rabbitmq -p 5672:567. -p 15672:15672 rabbitmq:3-management

अभिगमनपता: http://localhost:15672

पूर्वनिर्धारितः खातागुप्तशब्दः अस्ति: guest : guest

प्रशासकं रचयन्तु : १.

3. पङ्क्तिः एकैकं

 

P अस्माकं उत्पादकः अस्ति > मध्यपेटी एकः कतारः अस्ति, उपभोक्तृणा धारितस्य सन्देशबफरस्य प्रतिनिधित्वं करोति > C अस्माकं उपभोक्ता अस्ति

  1. 'use strict';
  2. const Controller = require('egg').Controller;
  3. /**
  4.  * 一对一队列演示
  5.  */
  6.  
  7. // 频道名称
  8. const queueName = 'hasone'
  9.  
  10. class UserController extends Controller {
  11.  
  12.   // 生成者
  13.   async send() {
  14.     // 1. 获取要发送的消息
  15.     const { msg } = this.ctx.query
  16.     // 2. 创建频道
  17.     const ch = await this.app.amqplib.createChannel();
  18.     // 3. 创建队列 durable 关闭持久化存储
  19.     await ch.assertQueue(queueName, { durable: false }  );
  20.     // 4. 发送消息
  21.     const ok = await ch.sendToQueue(queueName, Buffer.from(msg));
  22.     // 5. 关闭连接
  23.     await ch.close();
  24.  
  25.     this.ctx.body = ok;
  26.     this.ctx.status = 200;
  27.   }
  28.  
  29.   // 消费者
  30.   async work() {
  31.     // 1. 创建频道
  32.     const ch = await this.app.amqplib.createChannel();
  33.     // 2. 选择队列
  34.     await ch.assertQueue(queueName, { durable: false });
  35.     //3. 接收队列的消息
  36.     const resultMsg = await new Promise(resolve => ch.consume(queueName, msg => resolve(msg), { noAck: true }));
  37.  
  38.     // 4. 显示消息内容
  39.     if (resultMsg !== null) {
  40.       ch.ack(resultMsg);
  41.       await ch.close();
  42.  
  43.       const { content } = resultMsg;
  44.       this.status = 200;
  45.       this.ctx.body = { msg: content.toString() }
  46.  
  47.     } else {
  48.       this.ctx.body = '队列消费失败'
  49.       this.ctx.status = 500;
  50.     }
  51.   }
  52. }
  53.  
  54. module.exports = UserController;

4. पङ्क्तिः एक-बहुभ्यः

  1. 'use strict';
  2. const Controller = require('egg').Controller;
  3. /**
  4.  * 队列一对多演示
  5.  * 生产者 ---->  队列 ----> 消费者
  6. *                    ----> 消费者
  7.                      ----> 消费者
  8.  */
  9.  
  10. // 频道名称
  11. const queueName = 'hasMany'
  12.  
  13. class UserController extends Controller {
  14.  
  15.   // 生成者
  16.   async send() {
  17.     const { msg } = this.ctx.query;
  18.     //1. 创建频道
  19.     const ch = await this.app.amqplib.createChannel();
  20.     // 2. 创建队列 开启持久化存储
  21.     await ch.assertQueue(queueName, { durable: true });
  22.     // 3. 发送消息
  23.     let ok = null;
  24.     for(let i=0; i<50; i++) {
  25.       // 此时我们确信即使RabbitMQ重新启动，task_queue队列也不会丢失。现在我们需要将消息标记为持久性 - 通过使用持久性选项Channel.sendToQueue。
  26.       ok = await ch.sendToQueue(queueName, Buffer.from(msg+i), { persistent: true });
  27.     }
  28.     //4. 关闭连接
  29.     await ch.close();
  30.  
  31.     this.ctx.body = ok;
  32.     this.ctx.status = 200;
  33.   }
  34.  
  35.   // 消费者
  36.   async work1() {
  37.    // 1. 创建频道
  38.    const ch = await this.app.amqplib.createChannel();
  39.    //2. 选择队列
  40.    await ch.assertQueue(queueName, { durable: true });
  41.    // 3. 接收消息 noAck 关闭消息自动确认模式，需要手动 ack
  42.    const resultMsg = await new Promise(resolve => ch.consume(queueName, msg => {
  43.  
  44.     setTimeout(() => {
  45.       resolve(msg)
  46.      }, 500)
  47.  
  48.    }, { noAck: false }) );
  49.  
  50.    if (resultMsg !== null) {
  51.     const { content } = resultMsg;
  52.      //消费者发回ack（nowledgement）告诉RabbitMQ已收到，处理了特定消息，RabbitMQ可以自由删除它
  53.     ch.ack(resultMsg);
  54.     await ch.close();
  55.  
  56.     this.ctx.body = { work1: content.toString() };
  57.     this.ctx.status = 200;
  58.    } else {
  59.      this.ctx.body = '消费者1号失败'
  60.      this.ctx.status = 500
  61.    }
  62.  
  63.   }
  64.  
  65.   async work2() {
  66.      // 1. 创建频道
  67.    const ch = await this.app.amqplib.createChannel();
  68.    //2. 选择队列 RabbitMQ永远不会丢失我们的队列。为此，我们需要声明它是持久的
  69.    await ch.assertQueue(queueName, { durable: true });
  70.    // 3. 接收消息 noAck 开启自动确认模式
  71.    const resultMsg = await new Promise(resolve => ch.consume(queueName, msg => {
  72.     setTimeout(() => {
  73.       resolve(msg)
  74.      }, 1000)
  75.  
  76.    }, { noAck: false }) );
  77.  
  78.    if (resultMsg !== null) {
  79.     const { content } = resultMsg;
  80.     ch.ack(resultMsg);
  81.     await ch.close();
  82.  
  83.     this.ctx.body = { work2: content.toString() };
  84.     this.ctx.status = 200;
  85.    } else {
  86.      this.ctx.body = '消费者2号失败'
  87.      this.ctx.status = 500
  88.    }
  89.   }
  90.  
  91.   async work3() {
  92.      // 1. 创建频道
  93.    const ch = await this.app.amqplib.createChannel();
  94.    //2. 选择队列
  95.    await ch.assertQueue(queueName, { durable: true });
  96.    // 3. 接收消息 noAck 开启自动确认模式
  97.    const resultMsg = await new Promise(resolve => ch.consume(queueName, msg => {
  98.  
  99.     setTimeout(() => {
  100.       resolve(msg)
  101.      }, 1500)
  102.  
  103.  
  104.    }, { noAck: false }) );
  105.  
  106.  
  107.    if (resultMsg !== null) {
  108.     const { content } = resultMsg;
  109.     //消费者发回ack（nowledgement）告诉RabbitMQ已收到，处理了特定消息，RabbitMQ可以自由删除它
  110.     ch.ack(resultMsg);
  111.     await ch.close();
  112.  
  113.     this.ctx.body = { work3: content.toString() };
  114.     this.ctx.status = 200;
  115.    } else {
  116.      this.ctx.body = '消费者3号失败'
  117.      this.ctx.status = 500
  118.    }
  119.  
  120.   }
  121. }
  122.  
  123. module.exports = UserController;

व्यक्तिगत प्रोफाइल

सः ३० वर्षाणाम् अधिकं कालात् प्रौद्योगिक्याः शोधकार्यं कर्तुं समर्पितः अस्ति, तथा च जावा, लिनक्स, जावास्क्रिप्ट्, php, css इत्यादिषु विविधभाषासु प्रवीणः अस्ति, मुक्तस्रोतक्षेत्रे सः बहु योगदानं कृतवान् अस्ति विकासक दस्तावेजीकरणस्थानकं भविष्ये सन्दर्भार्थं प्रौद्योगिकीविकासे केचन विषयाः साझां कर्तुं सर्वे तत् पश्यन्तु

मम सम्पर्कसूचना

मेल