[TOC] ### 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; ```