工作队列

rabbitMQ 工作队列 轮询分发

>[danger] 简单队列是一对一的关系，一个生成者对应一个消费者，实际开发中，一般消费者是以业务相结合的，需要时间去处理业务，如果只有一个消费者，那么生产者就会积压很多消息，消费不出去

代码演示：

'use strict';
const Controller = require('egg').Controller;
/**
 * 队列一对多演示
 * 生产者 ---->  队列 ----> 消费者
*                    ----> 消费者
                     ----> 消费者
 */

// 频道名称
const queueName = 'hasMany'

class UserController extends Controller {

  // 生成者
  async send() {
    const { msg } = this.ctx.query;
    //1. 创建频道
    const ch = await this.app.amqplib.createChannel();
    // 2. 创建队列 开启持久化存储
    await ch.assertQueue(queueName, { durable: true });
    // 3. 发送消息
    let ok = null;
    for(let i=0; i<50; i++) {
      // 此时我们确信即使RabbitMQ重新启动，task_queue队列也不会丢失。现在我们需要将消息标记为持久性 - 通过使用持久性选项Channel.sendToQueue。
      ok = await ch.sendToQueue(queueName, Buffer.from(msg+i), { persistent: true });
    }
    //4. 关闭连接
    await ch.close();

    this.ctx.body = ok;
    this.ctx.status = 200;
  }

  // 消费者
  async work1() {
   // 1. 创建频道
   const ch = await this.app.amqplib.createChannel();
   //2. 选择队列
   await ch.assertQueue(queueName, { durable: true });
   // 3. 接收消息 noAck 关闭消息自动确认模式，需要手动 ack
   const resultMsg = await new Promise(resolve => ch.consume(queueName, msg => {

    setTimeout(() => {
      resolve(msg)
     }, 500)

   }, { noAck: false }) );

   if (resultMsg !== null) {
    const { content } = resultMsg;
     //消费者发回ack（nowledgement）告诉RabbitMQ已收到，处理了特定消息，RabbitMQ可以自由删除它
    ch.ack(resultMsg);
    await ch.close();

    this.ctx.body = { work1: content.toString() };
    this.ctx.status = 200;
   } else {
     this.ctx.body = '消费者1号失败'
     this.ctx.status = 500
   }

  }

  async work2() {
     // 1. 创建频道
   const ch = await this.app.amqplib.createChannel();
   //2. 选择队列 RabbitMQ永远不会丢失我们的队列。为此，我们需要声明它是持久的
   await ch.assertQueue(queueName, { durable: true });
   // 3. 接收消息 noAck 开启自动确认模式
   const resultMsg = await new Promise(resolve => ch.consume(queueName, msg => {
    setTimeout(() => {
      resolve(msg)
     }, 1000)

   }, { noAck: false }) );

   if (resultMsg !== null) {
    const { content } = resultMsg;
    ch.ack(resultMsg);
    await ch.close();

    this.ctx.body = { work2: content.toString() };
    this.ctx.status = 200;
   } else {
     this.ctx.body = '消费者2号失败'
     this.ctx.status = 500
   }
  }

  async work3() {
     // 1. 创建频道
   const ch = await this.app.amqplib.createChannel();
   //2. 选择队列
   await ch.assertQueue(queueName, { durable: true });
   // 3. 接收消息 noAck 开启自动确认模式
   const resultMsg = await new Promise(resolve => ch.consume(queueName, msg => {

    setTimeout(() => {
      resolve(msg)
     }, 1500)


   }, { noAck: false }) );


   if (resultMsg !== null) {
    const { content } = resultMsg;
    //消费者发回ack（nowledgement）告诉RabbitMQ已收到，处理了特定消息，RabbitMQ可以自由删除它
    ch.ack(resultMsg);
    await ch.close();

    this.ctx.body = { work3: content.toString() };
    this.ctx.status = 200;
   } else {
     this.ctx.body = '消费者3号失败'
     this.ctx.status = 500
   }

  }
}

module.exports = UserController;