fix(Consumer): 使用自己编写的Timer代替setTimeout

Author: leizongmin

lib/consumer.js

@@ -8,6 +8,7 @@
 
 const EventEmitter = require('events').EventEmitter;
 const utils = require('./utils');
+const Timer = require('./timer');
 
 class Consumer extends EventEmitter {
 
@@ -64,6 +65,8 @@ class Consumer extends EventEmitter {
     };
     this._heartbeatTid = setInterval(heartbeat, this._heartbeat * 1000);
 
+    this._timer = new Timer(1000);
+
     this._debug = utils.debug('consumer:' + this.name);
     this._debug('created: queue=%s, capacity=%s, redis=%j', this.queue, this.capacity, options.redis);
 
@@ -148,6 +151,8 @@ class Consumer extends EventEmitter {
 
     };
 
+    const timer = this._timer;
+
     class Message {
 
       constructor(producerName, msgId, expire, data, originData) {
@@ -177,10 +182,10 @@ class Consumer extends EventEmitter {
 
       checkTimeout(callback) {
         if (this._isDone) return;
-        setTimeout(() => {
+        timer.add(this.expire, () => {
           this.reject(new utils.MessageProcessingTimeoutError());
           callback && callback(null);
-        }, Date.now() - this.expire * 1000);
+        });
       }
 
     }
@@ -199,6 +204,7 @@ class Consumer extends EventEmitter {
     clearInterval(this._heartbeatTid);
     this._redisPull.end();
     this._redis.end();
+    this._timer.destroy();
     callback && callback(null);
   }
 

lib/timer.js

@@ -0,0 +1,55 @@
+'use strict';
+
+/**
+ * super-queue Timer
+ *
+ * @authro Zongmin Lei <leizongmin@gmail.com>
+ */
+
+const EventEmitter = require('events').EventEmitter;
+const utils = require('./utils');
+
+let counter = 0;
+
+class Timer {
+
+  constructor(interval) {
+
+    this.interval = interval || 1000;
+    this.list = [];
+    this._tid = setInterval(() => {
+      this.check();
+    }, this.interval);
+
+    this.id = counter++;
+    this._debug = utils.debug('timer:#' + this.id);
+    this._debug('created');
+
+  }
+
+  add(expire, callback) {
+    this.list.push({expire, callback});
+  }
+
+  check() {
+    const now = utils.secondTimestamp();
+    this.list.sort((a, b) => a.expire - b.expire);
+    let i = 0;
+    for (; i < this.list.length; i++) {
+      const item = this.list[i];
+      if (item.expire > now) break;
+    }
+    const expires = this.list.slice(0, i);
+    this.list = this.list.slice(i);
+    for (const item of expires) {
+      item.callback(item.expire);
+    }
+  }
+
+  destroy() {
+    clearInterval(this._tid);
+  }
+
+}
+
+module.exports = Timer;