redis storage 简单的实现及一些问题的细节

[luoyjx]

redis storage 问题

redis 无法存储 resolve 和 reject 在 redis 中。

目前解决方案

那么换了个思路，resolve 和 reject 还是在原有的作用域内，利用了 Server 类继承的 EventEmitter ，它的事件机制来进行 promise 的状态改变操作。

代码见:
https://github.com/luoyjx/tomqueue/blob/dev/lib/dispatcher.js#L84-L108

新问题

在worker处理能力不足时，dispatcher的内存会暴涨，这显然不是预期的效果，但是需要dispatcher确保消息发送到worker 到 反馈接收成功 期间，promise部分 和 listener 只能驻留在内存中。

若dispatcher或者调度器能根据dispatcher的负载来弹性伸缩worker数量的话，那内存问题也不是问题了，不过复杂度提升了。如果dispatcher来做这件事可能过于耦合了。

原有问题

不管是 memory 或是 redis 实现的 store ，在都会存在内存不可控的因素，主要还是跟 dispatcher 需要知道这条消息确实是被 worker 完整的接收到了而不仅仅是发送出去了有关。

疑问

这个 worker 是作为整个队列中的一个小部件然后消费者另外实现客户端，还是作为一个消费者的角色?

[luoyjx]

用工具分析了一下，发现这种基于事件的方式还是存在内存问题，有大量的listener堆积在堆中，由于上下文的循环引用，内存一直无法释放。

尝试在dispatcher内部做一个使用发送事务id的映射resolve、reject的缓存也可实现redis队列存储，并在发送后一秒做超时检测，若未收到reply并进行后续操作(使用事务id取到缓存并改变promise的状态，并移除这个缓存)，将主动移除缓存。

不知道这种方案会不会好点呢？

[luoyjx]

重新使用 事务id + 对象缓存实现。

constructor(options) {
    super(options);
    this.options = Object.assign({}, defaultOptions, this.options);

    ...

    // this object used to cache unReplied payload's
    // transaction id and resolve object and reject object
    this.uncommitedCache = {};

    ...
  }

handleReply(reply, workerId) {
    ...

    this.channels.shift(reply.channel).then((task) => {
      debug('shift :', reply.channel);
      let hasTransactionId = false;

      if (task && task.tId) {
        hasTransactionId = true;
      }

      if (reply.error) {
        hasTransactionId && this.uncommitedCache[task.tId] &&
          this.uncommitedCache[task.tId][1](reply.error);
      } else {
        hasTransactionId && this.uncommitedCache[task.tId] &&
          this.uncommitedCache[task.tId][0](reply);
      }

      // remove current cache
      hasTransactionId && delete this.uncommitedCache[task.tId];

      if (worker) {
        delete worker.channels[reply.channel];
        worker.pending -= 1;
      }
      this._dispatch(reply.channel);
    });
  }

send() {
      ...

      // Create a transaction id for marking this transaction.
      const tId = uuid.v4();

      this.channels.push(channel, { payload, tId }).then((length) => {
        if (length === 1) {
          this._dispatch(channel);
        }

        this.uncommitedCache[tId] = [resolve, reject, Date.now()];

        // remove uncommited cache when touching timeout
        // and retry dispatch payload
        setTimeout(() => {
          debug('[clean handler] check if transaction %s done => %s .', tId, !this.uncommitedCache[tId]);
          this.uncommitedCache[tId] && delete this.uncommitedCache[tId];

          // TODO: max retry option
          // resend payload
          this._dispatch(channel);
        }, this.options.timeout || DEFAULT_SEND_TIMEOUT);
      });

      ...
}

改进的部分：

1. 规避 listener 和 context 循环引用导致内存暴涨，在内存增长过快或超过阀值时，进行worker扩容，可缓慢将内存使用情况缓解，不是非常完美。
2. 超时进行缓存清除重发。