Add rabbitmq: consumer/producer implementation.

app.js

@@ -5,6 +5,8 @@ module.exports = app => {
   app.loader.loadToApp(path.join(app.config.baseDir, 'app/error'), 'error', {
     caseStyle: 'upper',
   });
+
+  loadConsumers(app);
   app.sessionStore = {
     async get(key) {
       const res = await app.redis.get(key);
@@ -21,3 +23,8 @@ module.exports = app => {
     },
   };
 };
+
+function loadConsumers(app) {
+  const unlockConsumer = new (require('./app/consumer/unlock_inventory'))();
+  unlockConsumer.init(app).catch(console.log);
+}

app/consumer/base_comsumer.js

@@ -0,0 +1,47 @@
+'use strict';
+
+const rmq = require('amqplib');
+
+class BaseConsumer {
+
+  static get ['QUEUE_NAME_PREF']() {
+    return 'mi_';
+  }
+
+  get ['queueName']() {
+    throw new Error('Please define your own queue name');
+  }
+
+  async initConn() {
+    const conn = await rmq.connect(this.app.config.rabbitmq.url);
+    this.conn = conn;
+  }
+
+  async initChannel() {
+    if (!this.conn) {
+      await this.initConn();
+    }
+    const channel = await this.conn.createChannel(this.queueName);
+    await channel.assertQueue(this.queueName, {
+      durable: true,
+    });
+    this.channel = channel;
+    return channel;
+  }
+
+  onMsg(msg) {
+    throw new Error('Please define your own message handler');
+  }
+
+  async init(app) {
+    if (!this.app) this.app = app;
+    if (!this.channel) {
+      await this.initChannel();
+    }
+    this.channel.consume(this.queueName, (msg)=>{this.onMsg(msg)});
+  }
+
+}
+
+module.exports = BaseConsumer;
+

app/consumer/unlock_inventory.js

@@ -0,0 +1,18 @@
+'use strict';
+
+const BaseConsumer = require('./base_comsumer');
+
+class UnlockInventory extends BaseConsumer {
+
+  get ['queueName']() {
+    return BaseConsumer.QUEUE_NAME_PREF + 'inventory_unlock';
+  }
+
+  onMsg(msg) {
+    console.log(JSON.parse(msg.content.toString('utf-8')));
+    this.channel.ack(msg);
+  }
+
+}
+
+module.exports = UnlockInventory;

app/controller/site.js

@@ -46,6 +46,8 @@ class SiteController extends Controller {
   }
 
   async index() {
+    await this.ctx.service.msgProducer.sendInventoryUnlockMsg(
+      { goodsId: '111', subOrderId: '2222' });
     this.ctx.body = {
       code: 0,
       data: {

app/service/inventory.js

@@ -39,7 +39,7 @@ class Inventory extends Service {
     return lockedAmount;
   }
 
-  async removeLocksByGoodsId(goodsId, subOrderId) {
+  async removeLocksByGoodsId(goodsId, subOrderId, amount) {
     const locksKey = Inventory.REDIS_GOODS_LOCKS_PREF + goodsId;
     const locksByOrderHsetKey = Inventory.REDIS_GOODS_LOCKS_BY_ORDER_PREF +
       goodsId;

app/service/msg_producer.js

@@ -0,0 +1,56 @@
+'use strict';
+
+const { Service } = require('egg');
+
+const rmq = require('amqplib');
+
+const QUEUE_NAME_PREF = 'mi_';
+
+class MsgProducer extends Service {
+
+  static get ['QUEUE_NAMES']() {
+    return {
+      INVENTORY_UNLOCK: QUEUE_NAME_PREF + 'inventory_unlock',
+    };
+  }
+
+  async initConn() {
+    const conn = await rmq.connect(this.app.config.rabbitmq.url);
+    this.conn = conn;
+  }
+
+  async initChannel(queueName) {
+    if (!this.conn) {
+      await this.initConn();
+    }
+    const channel = await this.conn.createChannel(queueName);
+    await channel.assertQueue(queueName, {
+      durable: true,
+    });
+    this.channels[queueName] = channel;
+    return channel;
+  }
+
+  async sendMsg(queueName, msg) {
+
+    if (!this.channels) {
+      this.channels = {};
+    }
+
+    let channel = this.channels[queueName];
+
+    if (!channel) {
+      channel = await this.initChannel(queueName);
+    }
+
+    await channel.sendToQueue(queueName, Buffer.from(msg));
+  }
+
+  async sendInventoryUnlockMsg(inventoryInfo) {
+    await this.sendMsg(MsgProducer.QUEUE_NAMES.INVENTORY_UNLOCK, JSON.stringify(inventoryInfo));
+  }
+
+}
+
+module.exports = MsgProducer;
+

config/config.default.js

@@ -58,5 +58,9 @@ module.exports = appInfo => {
     match: ['/api/v1/admin'],
   };
 
+  config.rabbitmq = {
+    url: 'amqp://localhost',
+  };
+
   return config;
 };