Redis操作场景下使用Promise.allSettled是否安全？

我正在编写一个涉及Redis操作的脚本，每分钟调用一次testFunction。这个函数会从Redis有序集合test_sorted_set中获取当前分钟之前的所有键（每个键对应另一个有序集合），然后并行调用processData和processSet来提升执行效率。但我担心会出问题——processSet会从成员有序集合中删除部分元素，而processData需要获取该有序集合的长度。

请问这种场景下用Promise.allSettled合适吗？还是应该改用两个独立的await？

testFunction 代码 #

const testFunction=async()=>{
  //ts for current minute
  let date = new Date();
  date.setUTCSeconds(0);
  date.setUTCMilliseconds(0);
  const currentTimestamp = date.getTime();

  //gets all keys till current minute (ZRANGE)
  let [keys, rangeError] = await cacheService.getDataInRange(
    'test_sorted_set',
    '-inf',
    currentTimestamp,
    'BYSCORE'
  );
  for (let member of keys) {
    const results = await Promise.allSettled([
      processData(member),
      processSet(member),
    ]);
    let processSetSuccess = true;
    for (let promiseResult of results) {
      const [_, processSetError] = promiseResult['value'];
      if (processSetError) {
        processSetSuccess = false;
        console.log(
          'Something went wrong while processing: ',
          processSetError
        );
      }
    }
    if (processSetSuccess) {
      //do something
    }
  }
}

processData 函数代码 #

const processData = async (member) => {
  //get length of member sorted set(zlen)
  const [count, countErr] = await cacheService.getSortedSetLength(member);
  if (countErr) {
    return [null, countErr];
  }

  let key = getKey(member);
  //3 get data from hash for particular key
  let [currentArr, hashError] = await cacheService.getFieldFromHash(
    "test_data",
    key,
  );
  if (hashError) {
    return [null, hashError];
  }
  let hashArray = [];
  currentArr = JSON.parse(currentArr);
  if (currentArr && currentArr.length > 0) {
    //array already present
    hashArray = await getNewArray(currentArr, count, timestamp);
  } else {
    //new entry
    hashArray = await getNewArray([], count, timestamp);
  }

  //4 Add updated array to hash
  const [addSuccess, addErr] = await cacheService.addToHash(
    "test_data",
    key,
    JSON.stringify(hashArray),
  );
  if (addErr) {
    return [null, addErr];
  }
  return [true, null];
};

processSet 函数代码 #

const processSet = async (member) => {
  //Remove all keys with 0 score
  const [removeSuccess, removeErr] = await cacheService.removeByScore(
    member,
    0,
    0,
  );

  let newSortedSetKey = getNextTimestamp(member);

  let newTS = parseInt(newSortedSetKey.split("_")[1]);
  //Store Difference between current and next set in a temporary sorted set
  const [_, diffErr] = await cacheService.storeSortedSetDifference(
    `${newSortedSetKey}_temp`,
    member,
    newSortedSetKey,
  );

  //Combine the temporary set and next set (union)
  const [destCount, unionErr] = await cacheService.unionStore(
    newSortedSetKey,
    newSortedSetKey,
    `${newSortedSetKey}_temp`,
  );

  if (destCount > 0) {
    //push to test_keys
    const [addKeysSuccess, addKeysError] = await cacheService.addToSortedSet(
      "sorted_set_1",
      newTS,
      newSortedSetKey,
    );
  }

  const [delSuccess, delError] = await cacheService.deleteDataStructure(
    `${newSortedSetKey}_temp`,
  );
  return [true, null];
};

Redis Service 核心函数代码 #

const getSortedSetLength = async (setKey) => {
    const [data, err] = await redisConnection.zcard(setKey);
    if (err) {
      console.log('REDIS ERROR');
      console.log(err);
      captureException(err);
    }
    return [data, err];
  };


  const removeByScore = async (key, minScore, maxScore) => {
    const [data, err] = await redisConnection.zremrangebyscore(key, minScore, maxScore);
    if (err) {
      console.log('REDIS ERROR');
      console.log(err);
      captureException(err);
    }
    return [data, err];
  };

绝对不能用Promise.allSettled并行执行这两个函数，必须改用顺序执行的两个await，原因如下：

1. 数据一致性风险：并行执行时，processSet里的zremrangebyscore（删除0分元素）可能在processData调用zcard获取集合长度之前完成，导致processData拿到的是删除后的长度，和你预期的「删除前的集合长度」不符，直接破坏业务逻辑的正确性。

2. Promise.allSettled解决不了时序问题：这个方法只是保证所有Promise都完成后再继续，不管成功失败，但它完全不控制两个Promise的执行顺序。并行执行的Redis操作谁先完成是不确定的，完全依赖Redis的响应速度，属于不可控的随机行为。

3. 正确的执行顺序：必须先调用processData，等它完成所有操作（尤其是拿到正确的集合长度）之后，再调用processSet执行删除等修改操作。示例代码如下：

   // 先执行processData，确保拿到修改前的长度
   const [dataRes, dataErr] = await processData(member);
   if (dataErr) {
     // 处理processData错误
     console.log('processData执行出错:', dataErr);
   }
   // 再执行processSet修改集合
   const [setRes, setErr] = await processSet(member);
   if (setErr) {
     // 处理processSet错误
     console.log('processSet执行出错:', setErr);
   }
   // 后续逻辑
   if (!dataErr && !setErr) {
     //do something
   }
   

4. 关于效率：虽然顺序执行会比并行慢一点，但这是保证数据正确性的必要代价。如果确实想优化，可以考虑在processSet中先获取集合长度并传递给processData，但这样会多一次Redis调用，且如果中间有其他外部修改仍存在风险，不如直接顺序执行可靠。

另外，你当前的Promise.allSettled错误处理逻辑也有问题：遍历结果时错误地把processData的返回值当成processSet的结果，会导致错误的异常判断，逻辑完全混乱。

内容的提问来源于stack exchange，提问作者Shardul Birje