SQS Consumer停止后仍接收消息的原因解析

我有一个Node.js应用，会监听云编排器发送的SIGTERM信号。关机时第一步先关闭SQS Consumer，再等待所有任务执行完成，但即使调用了停止操作，Consumer仍在处理消息。

相关代码 #

if (consumer != null) { // sanity check
  logger.log("Stopping SQS consumer...");
  try {
    consumer.stop();
    logger.log("SQS consumer stopped.");
  } catch (error: any) {
    logger.warn(`Error stopping SQS consumer during shutdown, this ain't good. error=${error}`);
  }
}

日志内容 #

{"name":"Index","hostname":"blah","pid":1,"level":"info","msg":"SQS consumer stopped.","time":"2023-03-21T22:58:13.418Z","v":0,"timestamp":"1679439493418"}

{"name":"Index","hostname":"blah","pid":1,"level":"info","msg":"Message received:  ...", "time":"2023-03-21T22:58:14.872Z","v":0,"timestamp":"1679439494872"}

两条日志的时间差为1.5秒，使用的sqs-consumer版本为5.6.0。

原因分析 #

1. stop()是异步操作未被等待：sqs-consumer 5.x版本中，consumer.stop()返回一个Promise，它仅触发停止流程，不会立即阻塞到Consumer完全停止。你的代码调用stop()后直接打印“已停止”日志，此时Consumer可能还在处理已拉取的消息，甚至还在进行SQS消息拉取请求。
2. SQS长轮询的特性：如果调用stop()前，Consumer已经发起了SQS的长轮询请求，这个请求不会被立即取消。当长轮询等待到新消息返回时，Consumer依然会触发消息接收回调，就会出现日志中“停止后仍收到消息”的情况。

修复建议 #

修改代码，等待stop()返回的Promise完成，确保Consumer真正停止后再执行后续逻辑：

if (consumer != null) { // sanity check
  logger.log("Stopping SQS consumer...");
  try {
    await consumer.stop(); // 等待停止操作完成
    logger.log("SQS consumer stopped.");
  } catch (error: any) {
    logger.warn(`Error stopping SQS consumer during shutdown, this ain't good. error=${error}`);
  }
}

注意：需要将SIGTERM信号的监听回调改为async函数，才能使用await关键字。

内容的提问来源于stack exchange，提问作者Ihor M.