javascript - 如何打破 SIGTERM 上的异步系列？

Question

假设我有以下场景 -

async.series(
  [
    function (cbi) {
      students.getAll('student', function (err, response) {
        if (err) {
          logger.error(err);
        }
        cbi(err, response);
      });
    },
    function (cbi) {
      students.deleteAll('student', function (err, response) {
        if (err) {
          logger.error(err);
        }
        cbi(err, response);
      });
    },
    function (cbi) {
      teachers.getAll('teacher', function (err, response) {
        if (err) {
          logger.error(err);
        }
        cbi(err, response);
      });
    },
    function (cbi) {
      teachers.deleteAll('teacher', function (err, response) {
        if (err) {
          logger.error(err);
        }
        cbi(err, response);
      });
    };
  ]
);

我希望在发送 SIGTERM 时进行优雅的清理。这是对所有学生或所有教师的清理，无论哪个正在进行，当信号发送时应该完成并且下一个不应该开始。

function (cbi) {
  students.getAll('student', function (err, response) {
    if (err || GLOBAL_VAR_SIGTERM === true) {
      logger.error(err);
    }
    cbi(err, response);
  });
}

我在想我应该设置一个全局变量来跟踪 SIGTERM 信号。

process.on('SIGTERM', function onSigterm () {
  GLOBAL_VAR_SIGTERM = true;
}

有没有更好的方法来打破异步系列来打破 SIGTERM 信号？

Answer 1

正如@adamrights 在 his answer 中指出的那样，代码中的主要问题是您没有使用真实的 err 第一个参数调用 cbi(err, response)，这对于停止 async 至关重要。 series 继续队列中的下一个任务。

现在您的代码应该可以工作了，但是您的代码中有一个重复模式:

function (cbi) {
  students.getAll('student', function (err, response) {
    // these 3 lines appear in every callback function
    if (GLOBAL_VAR_SIGTERM) err = new Error("SIGTERM: Aborting remaining tasks");
    if (err) logger.error(err);
    cbi(err, response);
    // end of repeat pattern
  });
}

传递给每个异步任务的回调总是做同样的三行事情。我们知道 DRY 规则，将重复模式提取到另一个函数中以尽可能重用它总是一个好主意。

因此，与其重复声明匿名函数，不如声明一个工厂函数。

function callbackFactory(cbi) {
  return function(err, response) {
    if (GLOBAL_VAR_SIGTERM) err = new Error("SIGTERM: Aborting remaining tasks");   
    if (err) logger.error(err);
    cbi(err, response);
  }
}

// use arrow function to write more concise code
async.series(
  [
    cbi => students.getAll('student', callbackFactory(cbi)),
    cbi => students.deleteAll('student', callbackFactory(cbi)),
    cbi => teachers.getAll('teacher', callbackFactory(cbi)),
    cbi => teachers.deleteAll('teacher', callbackFactory(cbi)),
  ]
);

---

进阶话题:使用装饰器处理横切关注点

让我们进一步探讨这个话题。显然，在收到 SIGTERM 时尽早中止是一个横切关注点，应该与业务逻辑分开。假设您的业务逻辑因任务而异:

async.series(
  [
    cbi => students.getAll('student', (err, response) => {
      if (err) {
        logger.error(err);
        return cbi(err);
      }
      updateStudentCount(response.data.length)  // <- extra work
      cbi(err, response);
    }),
    cbi => teachers.getAll('student', (err, response) => {
      if (err) {
        logger.error(err);
        return cbi(err);
      }
      updateTeacherCount(response.data.length)  // <- different extra work
      cbi(err, response);
    })
  ]
);

因为回调是变化的，所以很难像以前那样提取到工厂函数中。从这个 Angular 来说，我们最好注入(inject)每个任务的abort-early行为，方便编写正常的业务逻辑。

这就是装饰器模式派上用场的地方。但是全局变量并不是实现它的最佳工具，我们将使用事件监听器。

装饰器的基本界面如下:

// `task` will be things like `cbi => students.getAll('student', ... )`
function decorateTaskAbortEarly(task) {
  return (originalCbi) => {
    ...
    task(originalCbi)
  }
}

以下是我们的实现 list :

• 如果我们收到 SIGTERM，我们将调用 originalCbi
• 但是当我们没有收到 SIGTERM 时，originalCbi 仍然可以像正常情况一样在任何异步任务的回调中调用
• 如果 originalCbi 被调用一次，我们应该取消订阅 SIGTERM 以防止内存泄漏

实现:

function decorateTaskAbortEarly(task) {
  return (originalCbi) => {
    // subscribe to `SIGTERM`
    var listener = () => originalCbi(new Error("SIGTERM: Aborting remaining tasks"));
    process.once('SIGTERM', listener);

    var wrappedCbi = (err, response) => {
      // unsubscribe if `cbi` is called once
      process.off('SIGTERM', listener);
      return originalCbi(err, response);
    };
    // pass `cbi` through to `task`
    task(wrappedCbi);
  }
}

// Usage:
async.series(
  [
    cbi => students.getAll('student', (err, response) => {
      if (err) {
        logger.error(err);
        return cbi(err);
      }
      updateStudentCount(response.data.length)
      cbi(err, response);
    }),
    cbi => teachers.getAll('student', (err, response) => {
      if (err) {
        logger.error(err);
        return cbi(err);
      }
      updateTeacherCount(response.data.length)
      cbi(err, response);
    })
  ].map(decorateTaskAbortEarly)  // <--- nice API
);