RabbitMQ integration

2022-10-23 21:46:07 +02:00 · 2022-10-23 21:46:07 +02:00 · f5b84e338f
parent baff147435
commit f5b84e338f
7 changed files with 553 additions and 83 deletions
--- a/config.json
+++ b/config.json
@ -1,6 +1,6 @@
 {
  "debug": true,
-  "testMode": true,
+  "testMode": false,
  "db": {
    "host": "localhost",
    "port": 3306,
@ -8,6 +8,14 @@
    "password": "password",
    "database": "util"
  },
+  "consumerDb": {
+    "host": "localhost",
+    "port": 3306,
+    "user": "zongji",
+    "password": "password",
+    "database": "util"
+  },
+  "amqp": "amqp://user:password@localhost:5672",
  "includeEvents": [
    "rotate",
    "tablemap",
@ -16,9 +24,10 @@
    "deleterows"
  ],
  "pingInterval": 60,
-  "flushInterval": 10000,
+  "flushInterval": 5000,
  "queue": "orderRecalc",
  "addQuery": "INSERT INTO `hedera`.`orderRecalc` (`orderFk`) VALUES ?",
+  "recalcQuery": "CALL `hedera`.`order_recalc`(?)",
  "includeSchema": {
    "hedera": {
      "order": {
--- a/consumer.js
+++ b/consumer.js
@ -0,0 +1,137 @@
+
+const fs = require('fs');
+const path = require('path');
+const defaultConfig = require('./config.json');
+
+const mysql = require('mysql2/promise');
+const amqp = require('amqplib');
+require('colors');
+
+const config = Object.assign({}, defaultConfig);
+const localPath = path.join(__dirname, 'config.local.json');
+if (fs.existsSync(localPath)) {
+  const localConfig = require(localPath);
+  Object.assign(config, localConfig);
+}
+
+class Consumer {
+  constructor(config) {
+    this.config = config;
+  }
+
+  async start() {
+    if (this.config.testMode)
+      console.log('Test mode enabled, just logging queries to console.');
+
+    console.log('Starting process.');
+    await this.init();
+    console.log('Process started.');
+  }
+
+  async stop() {
+    console.log('Stopping process.');
+    await this.end();
+    console.log('Process stopped.');
+  }
+
+  async init() {
+    this.onErrorListener = err => this.onError(err);
+
+    this.db = await mysql.createConnection(this.config.consumerDb);
+    this.db.on('error', this.onErrorListener);
+
+    this.pingInterval = setInterval(
+      () => this.connectionPing(), this.config.pingInterval * 1000);
+
+    this.consumer = await amqp.connect(this.config.amqp);
+    this.channel = await this.consumer.createChannel();
+    this.channel.assertQueue(this.config.queue, {
+      durable: true
+    });
+    this.channel.consume(this.config.queue, msg => this.onConsume(msg));
+  }
+
+  async end(silent) {
+    clearInterval(this.pingInterval);
+
+    await this.consumer.close();
+
+    this.db.off('error', this.onErrorListener);
+    // FIXME: mysql2/promise bug, db.end() ends process
+    this.db.on('error', () => {});
+    try {
+      await this.db.end();
+    } catch (err) {
+      if (!silent)
+        console.error(err);
+    }
+  }
+
+  async connectionPing() {
+    this.debug('Ping', 'Sending ping to database.');
+    await this.db.ping();
+  }
+
+  async onConsume(msg) {
+    const fks = JSON.parse(msg.content.toString());
+    if (this.config.debug)
+      console.debug('RabbitMQ message'.blue, fks);
+
+    for (const fk of fks) {
+      if (this.config.debug)
+        console.debug('Query'.blue, this.config.recalcQuery.yellow);
+      await this.db.query(this.config.recalcQuery, fk);
+    }
+
+    this.channel.ack(msg);
+  }
+
+  async tryRestart() {
+    try {
+      await this.init();
+      console.log('Process restarted.');
+    } catch(err) {
+      setTimeout(() => this.tryRestart(), 30);
+    }
+  }
+
+  async onError(err) {
+    console.log(`Error: ${err.code}: ${err.message}`);
+    try {
+      await this.end(true);
+    } catch(e) {}
+
+    switch (err.code) {
+      case 'PROTOCOL_CONNECTION_LOST':
+      case 'ECONNRESET':
+      console.log('Trying to restart process.');
+      await this.tryRestart();
+      break;
+    default:
+      process.exit();
+    }
+  }
+
+  debug(namespace, message) {
+    if (this.config.debug)
+      console.debug(`${namespace}:`.blue, message.yellow);
+  }
+}
+
+let consumer;
+
+async function main() {
+  consumer = new Consumer(config)
+  await consumer.start();
+
+  process.on('SIGINT', async function() {
+    console.log('Got SIGINT.');
+    try {
+      await consumer.stop();
+    } catch (err) {
+      console.error(err);
+    }
+    process.exit();
+  });
+}
+main();
--- a/db-async.js
+++ b/db-async.js
@ -1,12 +1,13 @@
 const ZongJi = require('./zongji');
 const mysql = require('mysql2/promise');
+const amqp = require('amqplib');
+require('colors');

 const allEvents = new Set([
    'writerows',
    'updaterows',
    'deleterows'
  ]);
-const fks = new Set();

 module.exports = class DbAsync {
  constructor(config) {
@ -15,16 +16,7 @@ module.exports = class DbAsync {
    this.filename = null;
    this.position = null;
    this.schemaMap = new Map();
-  }
-
-  async start() {
-    if (this.config.testMode)
-      console.debug('Test mode enabled, just logging queries to console.');
-
-    console.log('Starting process.');
-
-    const db = await mysql.createConnection(this.config.db);
-    this.db = db;
+    this.fks = new Set();

    const includeSchema = {};
    for (const schemaName in this.config.includeSchema) {
@ -57,12 +49,52 @@ module.exports = class DbAsync {
      this.schemaMap.set(schemaName, tableMap);
    }
  
-    const opts = {
+    this.opts = {
      includeEvents: this.config.includeEvents,
      includeSchema
    };
-    this.opts = opts;
-  
+  }
+
+  async start() {
+    if (this.config.testMode)
+      console.log('Test mode enabled, just logging queries to console.');
+
+    console.log('Starting process.');
+    await this.init();
+    console.log('Process started.');
+  }
+
+  async stop() {
+    console.log('Stopping process.');
+    await this.end();
+    console.log('Process stopped.');
+  }
+
+  async init() {
+    this.debug('DbAsync', 'Initializing.');
+    this.onErrorListener = err => this.onError(err);
+
+    // DB connection
+
+    this.db = await mysql.createConnection(this.config.db);
+    this.db.on('error', this.onErrorListener);
+
+    // RabbitMQ
+
+    this.publisher = await amqp.connect(this.config.amqp);
+    this.channel = await this.publisher.createChannel();
+    this.channel.assertQueue(this.config.queue, {
+      durable: true
+    });
+
+    // Zongji
+
+    const zongji = new ZongJi(this.config.db);
+    this.zongji = zongji;
+
+    this.onBinlogListener = evt => this.onBinlog(evt);
+    zongji.on('binlog', this.onBinlogListener);
+
    const [res] = await this.db.query(
      'SELECT `logName`, `position` FROM `binlogQueue` WHERE code = ?',
      [this.config.queue]
@ -71,70 +103,122 @@ module.exports = class DbAsync {
      const [row] = res;
      this.filename = row.logName;
      this.position = row.position;
-      Object.assign(opts, {
+      Object.assign(this.opts, {
        filename: this.filename,
        position: this.position
      });
    } else
-      opts.startAtEnd = true;
+      this.opts.startAtEnd = true;

-    await this.startZongji();
-  }
+    this.debug('Zongji', 'Starting.');
+    await new Promise((resolve, reject) => {
+      const onReady = () => {
+        zongji.off('error', onError);
+        resolve();
+      };
+      const onError = err => {
+        this.zongji = null;
+        zongji.off('ready', onReady);
+        zongji.off('binlog', this.onBinlogListener);
+        reject(err);
+      }

-  async stop() {
-    await this.stopZongji();
-    await this.db.end();
-  }
+      zongji.once('ready', onReady);
+      zongji.once('error',  onError);
+      zongji.start(this.opts);
+    });
+    this.debug('Zongji', 'Started.');

-  async startZongji() {
-    const zongji = new ZongJi(this.config.db);
-    this.zongji = zongji;
-
-    zongji.on('ready', () => this.onReady());
-    zongji.on('stopped', () => this.onStopped());
-    zongji.on('error', err => this.onError(err));
-    zongji.on('binlog', evt => this.onBinlog(evt));
-    zongji.start(this.opts);
-  }
-
-  async stopZongji() {
-    console.debug('Stopping Zongji.');
-    this.running = false;
-    clearInterval(this.flushInterval);
-    clearInterval(this.pingInterval);
-    this.zongji.stop();
-  }
-
-  async restartZongji() {
-    console.debug('Restaring Zongji.');
-    await this.stopZongji();
-    setTimeout(() => this.startZongji(this.opts), 1000);
-  }
-
-  onReady() {
-    this.running = true;
+    this.zongji.on('error', this.onErrorListener);

    this.flushInterval = setInterval(
      () => this.flushQueue(), this.config.flushInterval);
    this.pingInterval = setInterval(
      () => this.connectionPing(), this.config.pingInterval * 1000);

-    console.debug('Zongji ready.');
+    // Summary
+
+    this.running = true;
+    this.debug('DbAsync', 'Initialized.');
  }

-  onStopped() {
-    console.debug('Zongji stopped.');
+  async end(silent) {
+    const zongji = this.zongji;
+    if (!zongji) return;
+
+    this.debug('DbAsync', 'Ending.');
+
+    // Zongji
+
+    clearInterval(this.flushInterval);
+    clearInterval(this.pingInterval);
+    zongji.off('binlog', this.onBinlogListener);
+    zongji.off('error', this.onErrorListener);
+    this.zongji = null;
+    this.running = false;
+
+    this.debug('Zongji', 'Stopping.');
+    // FIXME: Cannot call Zongji.stop(), it doesn't wait to end connection
+    zongji.connection.destroy(() => {
+      console.log('zongji.connection.destroy');
+    });
+    await new Promise(resolve => {
+      zongji.ctrlConnection.query('KILL ' + zongji.connection.threadId,
+      err => {
+        if (err && !silent)
+          console.error(err);
+        resolve();
+      });
+    });
+    zongji.ctrlConnection.destroy(() => {
+      console.log('zongji.ctrlConnection.destroy');
+    });
+    zongji.emit('stopped');
+    this.debug('Zongji', 'Stopped.');
+
+    // RabbitMQ
+
+    await this.publisher.close();
+
+    // DB connection
+
+    this.db.off('error', this.onErrorListener);
+    // FIXME: mysql2/promise bug, db.end() ends process
+    this.db.on('error', () => {});
+    try {
+      await this.db.end();
+    } catch (err) {
+      if (!silent)
+        console.error(err);
+    }
+
+    // Summary
+
+    this.debug('DbAsync', 'Ended.');
+  }
+
+  async tryRestart() {
+    try {
+      await this.init();
+      console.log('Process restarted.');
+    } catch(err) {
+      setTimeout(() => this.tryRestart(), 30);
+    }
  }

  async onError(err) {
    console.log(`Error: ${err.code}: ${err.message}`);
+    try {
+      await this.end(true);
+    } catch(e) {}
+
    switch (err.code) {
      case 'PROTOCOL_CONNECTION_LOST':
      case 'ECONNRESET':
-      await this.restartZongji();
+      console.log('Trying to restart process.');
+      await this.tryRestart();
      break;
    default:
-      await this.stop();
      process.exit();
    }
  }
@ -163,6 +247,8 @@ module.exports = class DbAsync {
    let column;
    const rows = evt.rows;

+    const fks = new Set();
+
    if (eventName === 'updaterows') {
      if (tableInfo.columns !== true) {
        let changes = false;
@ -187,6 +273,13 @@ module.exports = class DbAsync {
        fks.add(row[tableInfo.fk]);
    }

+    if (fks.size) {
+      const data = JSON.stringify(Array.from(fks));
+      this.channel.sendToQueue(this.config.queue,
+        Buffer.from(data));
+      this.debug('Queued', data);
+    }
+
    const row = eventName === 'updaterows'
      ? rows[0].after
      : rows[0];
@ -202,40 +295,38 @@ module.exports = class DbAsync {
    }

    this.position = evt.nextPosition;
+    this.flushed = false;
  }

  async flushQueue() {
-    if (!this.running) return;
-    console.log('==========================================================');
-    console.log('Flush:', `filename: ${this.filename}`, `position: ${this.position}`);
-    console.log(fks);
-    if (!fks.size) return;
-  
-    const ids = [];
-    for (const fk of fks) ids.push([fk]);
+    if (this.flushed) return;
+    this.debug('Flush', `filename: ${this.filename}, position: ${this.position}`);
  
    const replaceQuery =
      'REPLACE INTO `binlogQueue`  SET `code` = ?, `logName` = ?, `position` = ?';
-  
-    if (this.config.testMode) {
-      console.debug(this.config.addQuery);
-      console.debug(replaceQuery);
-    } else {
-      await this.db.query(this.config.addQuery, [ids]);
+    if (!this.config.testMode)
      await this.db.query(replaceQuery, [this.config.queue, this.filename, this.position]);
-    }
-    fks.clear();
-    console.log('==========================================================');
+
+    this.flushed = true;
  }

  async connectionPing() {
-    if (!this.running) return;
-    if (this.config.debug)
-      console.debug('Sending ping to database.')
-    this.zongji.connection.ping();
-    this.zongji.ctrlConnection.ping();
+    this.debug('Ping', 'Sending ping to database.');
+
+    // FIXME: Should Zongji.connection be pinged?
+    await new Promise((resolve, reject) => {
+      this.zongji.ctrlConnection.ping(err => {
+        if (err) return reject(err);
+        resolve();
+      });
+    })
    await this.db.ping();
  }
+
+  debug(namespace, message) {
+    if (this.config.debug)
+      console.debug(`${namespace}:`.blue, message.yellow);
+  }
 }

 function equals(a, b) {
--- a/index.js
+++ b/index.js
@ -14,15 +14,17 @@ if (fs.existsSync(localPath)) {
 let dbAsync;

 async function main() {
-
  dbAsync = new DbAsync(config)
  await dbAsync.start();
-  
+
  process.on('SIGINT', async function() {
    console.log('Got SIGINT.');
-    await dbAsync.stop();
+    try {
+      await dbAsync.stop();
+    } catch (err) {
+      console.error(err);
+    }
    process.exit();
  });
 }
-
 main();
--- a/package-lock.json
+++ b/package-lock.json
@ -1,15 +1,78 @@
 {
-  "name": "binlog",
+  "name": "db-async",
  "lockfileVersion": 2,
  "requires": true,
  "packages": {
    "": {
      "dependencies": {
+        "amqplib": "^0.10.3",
+        "colors": "^1.4.0",
        "mysql2": "^2.3.3",
        "zongji": "file:../zongji"
      }
    },
    "../zongji": {},
+    "node_modules/@acuminous/bitsyntax": {
+      "version": "0.1.2",
+      "resolved": "https://registry.npmjs.org/@acuminous/bitsyntax/-/bitsyntax-0.1.2.tgz",
+      "integrity": "sha512-29lUK80d1muEQqiUsSo+3A0yP6CdspgC95EnKBMi22Xlwt79i/En4Vr67+cXhU+cZjbti3TgGGC5wy1stIywVQ==",
+      "dependencies": {
+        "buffer-more-ints": "~1.0.0",
+        "debug": "^4.3.4",
+        "safe-buffer": "~5.1.2"
+      },
+      "engines": {
+        "node": ">=0.8"
+      }
+    },
+    "node_modules/amqplib": {
+      "version": "0.10.3",
+      "resolved": "https://registry.npmjs.org/amqplib/-/amqplib-0.10.3.tgz",
+      "integrity": "sha512-UHmuSa7n8vVW/a5HGh2nFPqAEr8+cD4dEZ6u9GjP91nHfr1a54RyAKyra7Sb5NH7NBKOUlyQSMXIp0qAixKexw==",
+      "dependencies": {
+        "@acuminous/bitsyntax": "^0.1.2",
+        "buffer-more-ints": "~1.0.0",
+        "readable-stream": "1.x >=1.1.9",
+        "url-parse": "~1.5.10"
+      },
+      "engines": {
+        "node": ">=10"
+      }
+    },
+    "node_modules/buffer-more-ints": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/buffer-more-ints/-/buffer-more-ints-1.0.0.tgz",
+      "integrity": "sha512-EMetuGFz5SLsT0QTnXzINh4Ksr+oo4i+UGTXEshiGCQWnsgSs7ZhJ8fzlwQ+OzEMs0MpDAMr1hxnblp5a4vcHg=="
+    },
+    "node_modules/colors": {
+      "version": "1.4.0",
+      "resolved": "https://registry.npmjs.org/colors/-/colors-1.4.0.tgz",
+      "integrity": "sha512-a+UqTh4kgZg/SlGvfbzDHpgRu7AAQOmmqRHJnxhRZICKFUT91brVhNNt58CMWU9PsBbv3PDCZUHbVxuDiH2mtA==",
+      "engines": {
+        "node": ">=0.1.90"
+      }
+    },
+    "node_modules/core-util-is": {
+      "version": "1.0.3",
+      "resolved": "https://registry.npmjs.org/core-util-is/-/core-util-is-1.0.3.tgz",
+      "integrity": "sha512-ZQBvi1DcpJ4GDqanjucZ2Hj3wEO5pZDS89BWbkcrvdxksJorwUDDZamX9ldFkp9aw2lmBDLgkObEA4DWNJ9FYQ=="
+    },
+    "node_modules/debug": {
+      "version": "4.3.4",
+      "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.4.tgz",
+      "integrity": "sha512-PRWFHuSU3eDtQJPvnNY7Jcket1j0t5OuOsFzPPzsekD52Zl8qUfFIPEiswXqIvHWGVHOgX+7G/vCNNhehwxfkQ==",
+      "dependencies": {
+        "ms": "2.1.2"
+      },
+      "engines": {
+        "node": ">=6.0"
+      },
+      "peerDependenciesMeta": {
+        "supports-color": {
+          "optional": true
+        }
+      }
+    },
    "node_modules/denque": {
      "version": "2.1.0",
      "resolved": "https://registry.npmjs.org/denque/-/denque-2.1.0.tgz",
@ -37,11 +100,21 @@
        "node": ">=0.10.0"
      }
    },
+    "node_modules/inherits": {
+      "version": "2.0.4",
+      "resolved": "https://registry.npmjs.org/inherits/-/inherits-2.0.4.tgz",
+      "integrity": "sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ=="
+    },
    "node_modules/is-property": {
      "version": "1.0.2",
      "resolved": "https://registry.npmjs.org/is-property/-/is-property-1.0.2.tgz",
      "integrity": "sha512-Ks/IoX00TtClbGQr4TWXemAnktAQvYB7HzcCxDGqEZU6oCmb2INHuOoKxbtR+HFkmYWBKv/dOZtGRiAjDhj92g=="
    },
+    "node_modules/isarray": {
+      "version": "0.0.1",
+      "resolved": "https://registry.npmjs.org/isarray/-/isarray-0.0.1.tgz",
+      "integrity": "sha512-D2S+3GLxWH+uhrNEcoh/fnmYeP8E8/zHl644d/jdA0g2uyXvy3sb0qxotE+ne0LtccHknQzWwZEzhak7oJ0COQ=="
+    },
    "node_modules/long": {
      "version": "4.0.0",
      "resolved": "https://registry.npmjs.org/long/-/long-4.0.0.tgz",
@ -58,6 +131,11 @@
        "node": ">=10"
      }
    },
+    "node_modules/ms": {
+      "version": "2.1.2",
+      "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.2.tgz",
+      "integrity": "sha512-sGkPx+VjMtmA6MX27oA4FBFELFCZZ4S4XqeGOXCv68tT+jb3vk/RyaKWP0PTKyWtmLSM0b+adUTEvbs1PEaH2w=="
+    },
    "node_modules/mysql2": {
      "version": "2.3.3",
      "resolved": "https://registry.npmjs.org/mysql2/-/mysql2-2.3.3.tgz",
@ -106,6 +184,32 @@
      "resolved": "https://registry.npmjs.org/pseudomap/-/pseudomap-1.0.2.tgz",
      "integrity": "sha512-b/YwNhb8lk1Zz2+bXXpS/LK9OisiZZ1SNsSLxN1x2OXVEhW2Ckr/7mWE5vrC1ZTiJlD9g19jWszTmJsB+oEpFQ=="
    },
+    "node_modules/querystringify": {
+      "version": "2.2.0",
+      "resolved": "https://registry.npmjs.org/querystringify/-/querystringify-2.2.0.tgz",
+      "integrity": "sha512-FIqgj2EUvTa7R50u0rGsyTftzjYmv/a3hO345bZNrqabNqjtgiDMgmo4mkUjd+nzU5oF3dClKqFIPUKybUyqoQ=="
+    },
+    "node_modules/readable-stream": {
+      "version": "1.1.14",
+      "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-1.1.14.tgz",
+      "integrity": "sha512-+MeVjFf4L44XUkhM1eYbD8fyEsxcV81pqMSR5gblfcLCHfZvbrqy4/qYHE+/R5HoBUT11WV5O08Cr1n3YXkWVQ==",
+      "dependencies": {
+        "core-util-is": "~1.0.0",
+        "inherits": "~2.0.1",
+        "isarray": "0.0.1",
+        "string_decoder": "~0.10.x"
+      }
+    },
+    "node_modules/requires-port": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/requires-port/-/requires-port-1.0.0.tgz",
+      "integrity": "sha512-KigOCHcocU3XODJxsu8i/j8T9tzT4adHiecwORRQ0ZZFcp7ahwXuRU1m+yuO90C5ZUyGeGfocHDI14M3L3yDAQ=="
+    },
+    "node_modules/safe-buffer": {
+      "version": "5.1.2",
+      "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.1.2.tgz",
+      "integrity": "sha512-Gd2UZBJDkXlY7GbJxfsE8/nvKkUEU1G38c1siN6QP6a9PT9MmHB8GnpscSmMJSoF8LOIrt8ud/wPtojys4G6+g=="
+    },
    "node_modules/safer-buffer": {
      "version": "2.1.2",
      "resolved": "https://registry.npmjs.org/safer-buffer/-/safer-buffer-2.1.2.tgz",
@ -124,6 +228,20 @@
        "node": ">= 0.6"
      }
    },
+    "node_modules/string_decoder": {
+      "version": "0.10.31",
+      "resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-0.10.31.tgz",
+      "integrity": "sha512-ev2QzSzWPYmy9GuqfIVildA4OdcGLeFZQrq5ys6RtiuF+RQQiZWr8TZNyAcuVXyQRYfEO+MsoB/1BuQVhOJuoQ=="
+    },
+    "node_modules/url-parse": {
+      "version": "1.5.10",
+      "resolved": "https://registry.npmjs.org/url-parse/-/url-parse-1.5.10.tgz",
+      "integrity": "sha512-WypcfiRhfeUP9vvF0j6rw0J3hrWrw6iZv3+22h6iRMJ/8z1Tj6XfLP4DsUix5MhMPnXpiHDoKyoZ/bdCkwBCiQ==",
+      "dependencies": {
+        "querystringify": "^2.1.1",
+        "requires-port": "^1.0.0"
+      }
+    },
    "node_modules/yallist": {
      "version": "4.0.0",
      "resolved": "https://registry.npmjs.org/yallist/-/yallist-4.0.0.tgz",
@ -135,6 +253,50 @@
    }
  },
  "dependencies": {
+    "@acuminous/bitsyntax": {
+      "version": "0.1.2",
+      "resolved": "https://registry.npmjs.org/@acuminous/bitsyntax/-/bitsyntax-0.1.2.tgz",
+      "integrity": "sha512-29lUK80d1muEQqiUsSo+3A0yP6CdspgC95EnKBMi22Xlwt79i/En4Vr67+cXhU+cZjbti3TgGGC5wy1stIywVQ==",
+      "requires": {
+        "buffer-more-ints": "~1.0.0",
+        "debug": "^4.3.4",
+        "safe-buffer": "~5.1.2"
+      }
+    },
+    "amqplib": {
+      "version": "0.10.3",
+      "resolved": "https://registry.npmjs.org/amqplib/-/amqplib-0.10.3.tgz",
+      "integrity": "sha512-UHmuSa7n8vVW/a5HGh2nFPqAEr8+cD4dEZ6u9GjP91nHfr1a54RyAKyra7Sb5NH7NBKOUlyQSMXIp0qAixKexw==",
+      "requires": {
+        "@acuminous/bitsyntax": "^0.1.2",
+        "buffer-more-ints": "~1.0.0",
+        "readable-stream": "1.x >=1.1.9",
+        "url-parse": "~1.5.10"
+      }
+    },
+    "buffer-more-ints": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/buffer-more-ints/-/buffer-more-ints-1.0.0.tgz",
+      "integrity": "sha512-EMetuGFz5SLsT0QTnXzINh4Ksr+oo4i+UGTXEshiGCQWnsgSs7ZhJ8fzlwQ+OzEMs0MpDAMr1hxnblp5a4vcHg=="
+    },
+    "colors": {
+      "version": "1.4.0",
+      "resolved": "https://registry.npmjs.org/colors/-/colors-1.4.0.tgz",
+      "integrity": "sha512-a+UqTh4kgZg/SlGvfbzDHpgRu7AAQOmmqRHJnxhRZICKFUT91brVhNNt58CMWU9PsBbv3PDCZUHbVxuDiH2mtA=="
+    },
+    "core-util-is": {
+      "version": "1.0.3",
+      "resolved": "https://registry.npmjs.org/core-util-is/-/core-util-is-1.0.3.tgz",
+      "integrity": "sha512-ZQBvi1DcpJ4GDqanjucZ2Hj3wEO5pZDS89BWbkcrvdxksJorwUDDZamX9ldFkp9aw2lmBDLgkObEA4DWNJ9FYQ=="
+    },
+    "debug": {
+      "version": "4.3.4",
+      "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.4.tgz",
+      "integrity": "sha512-PRWFHuSU3eDtQJPvnNY7Jcket1j0t5OuOsFzPPzsekD52Zl8qUfFIPEiswXqIvHWGVHOgX+7G/vCNNhehwxfkQ==",
+      "requires": {
+        "ms": "2.1.2"
+      }
+    },
    "denque": {
      "version": "2.1.0",
      "resolved": "https://registry.npmjs.org/denque/-/denque-2.1.0.tgz",
@ -156,11 +318,21 @@
        "safer-buffer": ">= 2.1.2 < 3.0.0"
      }
    },
+    "inherits": {
+      "version": "2.0.4",
+      "resolved": "https://registry.npmjs.org/inherits/-/inherits-2.0.4.tgz",
+      "integrity": "sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ=="
+    },
    "is-property": {
      "version": "1.0.2",
      "resolved": "https://registry.npmjs.org/is-property/-/is-property-1.0.2.tgz",
      "integrity": "sha512-Ks/IoX00TtClbGQr4TWXemAnktAQvYB7HzcCxDGqEZU6oCmb2INHuOoKxbtR+HFkmYWBKv/dOZtGRiAjDhj92g=="
    },
+    "isarray": {
+      "version": "0.0.1",
+      "resolved": "https://registry.npmjs.org/isarray/-/isarray-0.0.1.tgz",
+      "integrity": "sha512-D2S+3GLxWH+uhrNEcoh/fnmYeP8E8/zHl644d/jdA0g2uyXvy3sb0qxotE+ne0LtccHknQzWwZEzhak7oJ0COQ=="
+    },
    "long": {
      "version": "4.0.0",
      "resolved": "https://registry.npmjs.org/long/-/long-4.0.0.tgz",
@ -174,6 +346,11 @@
        "yallist": "^4.0.0"
      }
    },
+    "ms": {
+      "version": "2.1.2",
+      "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.2.tgz",
+      "integrity": "sha512-sGkPx+VjMtmA6MX27oA4FBFELFCZZ4S4XqeGOXCv68tT+jb3vk/RyaKWP0PTKyWtmLSM0b+adUTEvbs1PEaH2w=="
+    },
    "mysql2": {
      "version": "2.3.3",
      "resolved": "https://registry.npmjs.org/mysql2/-/mysql2-2.3.3.tgz",
@ -218,6 +395,32 @@
      "resolved": "https://registry.npmjs.org/pseudomap/-/pseudomap-1.0.2.tgz",
      "integrity": "sha512-b/YwNhb8lk1Zz2+bXXpS/LK9OisiZZ1SNsSLxN1x2OXVEhW2Ckr/7mWE5vrC1ZTiJlD9g19jWszTmJsB+oEpFQ=="
    },
+    "querystringify": {
+      "version": "2.2.0",
+      "resolved": "https://registry.npmjs.org/querystringify/-/querystringify-2.2.0.tgz",
+      "integrity": "sha512-FIqgj2EUvTa7R50u0rGsyTftzjYmv/a3hO345bZNrqabNqjtgiDMgmo4mkUjd+nzU5oF3dClKqFIPUKybUyqoQ=="
+    },
+    "readable-stream": {
+      "version": "1.1.14",
+      "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-1.1.14.tgz",
+      "integrity": "sha512-+MeVjFf4L44XUkhM1eYbD8fyEsxcV81pqMSR5gblfcLCHfZvbrqy4/qYHE+/R5HoBUT11WV5O08Cr1n3YXkWVQ==",
+      "requires": {
+        "core-util-is": "~1.0.0",
+        "inherits": "~2.0.1",
+        "isarray": "0.0.1",
+        "string_decoder": "~0.10.x"
+      }
+    },
+    "requires-port": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/requires-port/-/requires-port-1.0.0.tgz",
+      "integrity": "sha512-KigOCHcocU3XODJxsu8i/j8T9tzT4adHiecwORRQ0ZZFcp7ahwXuRU1m+yuO90C5ZUyGeGfocHDI14M3L3yDAQ=="
+    },
+    "safe-buffer": {
+      "version": "5.1.2",
+      "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.1.2.tgz",
+      "integrity": "sha512-Gd2UZBJDkXlY7GbJxfsE8/nvKkUEU1G38c1siN6QP6a9PT9MmHB8GnpscSmMJSoF8LOIrt8ud/wPtojys4G6+g=="
+    },
    "safer-buffer": {
      "version": "2.1.2",
      "resolved": "https://registry.npmjs.org/safer-buffer/-/safer-buffer-2.1.2.tgz",
@ -233,6 +436,20 @@
      "resolved": "https://registry.npmjs.org/sqlstring/-/sqlstring-2.3.3.tgz",
      "integrity": "sha512-qC9iz2FlN7DQl3+wjwn3802RTyjCx7sDvfQEXchwa6CWOx07/WVfh91gBmQ9fahw8snwGEWU3xGzOt4tFyHLxg=="
    },
+    "string_decoder": {
+      "version": "0.10.31",
+      "resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-0.10.31.tgz",
+      "integrity": "sha512-ev2QzSzWPYmy9GuqfIVildA4OdcGLeFZQrq5ys6RtiuF+RQQiZWr8TZNyAcuVXyQRYfEO+MsoB/1BuQVhOJuoQ=="
+    },
+    "url-parse": {
+      "version": "1.5.10",
+      "resolved": "https://registry.npmjs.org/url-parse/-/url-parse-1.5.10.tgz",
+      "integrity": "sha512-WypcfiRhfeUP9vvF0j6rw0J3hrWrw6iZv3+22h6iRMJ/8z1Tj6XfLP4DsUix5MhMPnXpiHDoKyoZ/bdCkwBCiQ==",
+      "requires": {
+        "querystringify": "^2.1.1",
+        "requires-port": "^1.0.0"
+      }
+    },
    "yallist": {
      "version": "4.0.0",
      "resolved": "https://registry.npmjs.org/yallist/-/yallist-4.0.0.tgz",
--- a/package.json
+++ b/package.json
@ -1,5 +1,7 @@
 {
  "dependencies": {
+    "amqplib": "^0.10.3",
+    "colors": "^1.4.0",
    "mysql2": "^2.3.3",
    "zongji": "file:../zongji"
  }
--- a/run-queue.sh
+++ b/run-queue.sh
@ -0,0 +1,12 @@
+#!/bin/bash
+
+docker rm -f some-rabbit
+
+docker run \
+    -d --hostname my-rabbit \
+    --name some-rabbit \
+    -e RABBITMQ_DEFAULT_USER=user \
+    -e RABBITMQ_DEFAULT_PASS=password \
+    -p 5672:5672 \
+    -p 8080:15672 \
+    rabbitmq:3-management