技術共有

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/egg-amqplib

制御層:

  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. };

参照する：egg-amqplib/test/fixtures/apps/amqplib-test/app/controller/home.js マスター · zubinzhang/egg-amqplib · GitHub

2. Rabbitmq のインストール docker を使用して RabbitMQ をインストールできます。

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

アクセスアドレス：http://localhost:15672

デフォルトのアカウントパスワードは次のとおりです: guest : guest

管理者の作成:

3. キュー: 1 対 1

 

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 対多

  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;

個人プロフィール

彼は 30 年以上テクノロジーの研究に専念しており、java、linux、javascript、php、css などのさまざまな言語に堪能であり、オープンソース分野で多くの貢献を行っています。将来の参考のために技術開発におけるいくつかの問題を共有する開発者ドキュメント ステーション。

プライベートな連絡先の最初の情報

送料