node.js - 无法使用 setTimeout 解决 promise

Question

我正在编写一个函数来使用 Node.js 写入 AWS Kinesis firehose。写入 firehose 流的 AWS 函数调用返回“错误”或“数据”，具体取决于操作结果。如果出现错误，对于特定的错误代码，我需要使用exponentialBackoff重试相同的请求。我使用 setTimeOut 为每次后续重试以不同的时间触发相同的方法，但看起来每当重试时，我都无法正确解析，并且我的测试失败，提示“错误:超出 5000 毫秒超时。对于异步测试和 Hook ，确保调用“done()”；如果返回 Promise，请确保它解析。”

async function batchWrite(records,firehose,retry = 0){
    var readingObjects = getReadings(records);
    var params = {
        DeliveryStreamName: process.env.KINESIS_FIREHOSE_STREAM_DELIVERY,
        Records: readingObjects
    };

    return await new Promise(function(resolve,reject){
        firehose.putRecordBatch(params,function(error,data){
            if(error){
                if(error.code == 'ServiceUnavailableException' && retry < retries.length){
                    console.log('retryCount=',retry);
                    setTimeout(batchWrite,retries[retry],records,firehose,retry+1);
                    console.log('setTimeout',retry);
                }
                else{
                    // console.log('Error',error);
                    console.log('resolving');
                    resolve(error);
                }
            }
            else{
                if(data.FailedPutCount > 0){
                    //colect the RequestResponses which are not processed.
                    //index of those records is same as in request
                    //process those faied records again
                    console.log('Error',data);
                }
                resolve(data);
            }
        });
    });
}

对于上面的函数，在测试用例中，我通过返回具有 ServiceUnavailableException 的响应来模拟错误情况，以便重试，但在所有重试之后，测试超时。我需要更改什么才能使其正常工作吗？

Answer 1

当您调用 setTimeout() 进行重试时，您创建并返回的原始 Promise 永远不会被解析。因此，调用者永远看不到已解决的 promise 。您需要将新的 promise 链接到前一个 promise 。而且，您不能直接使用 setTimeout() 来做到这一点，但您可以将其包装在自己的 promise 中，然后将其链接到原始 promise 。此外，如果您将 putRecordBatch promise 为最低级别并使用 promise 完成所有控制流程，而不是 promise 和回调的混合，那么代码会更易于管理。

具体操作方法如下:

// utility function to return a promise that is resolved after a setTimeout()
function delay(t, v) {
    return new Promise(resolve => {
        setTimeout(resolve, t, v);
    });
}

const promisify = require('util').promisify;

function batchWrite(records,firehose,retry = 0){
    var readingObjects = getReadings(records);
    var params = {
        DeliveryStreamName: process.env.KINESIS_FIREHOSE_STREAM_DELIVERY,
        Records: readingObjects
    };

    // make a promisified version of firehose.putRecordBatch
    firehose.putRecordBatchP = promisify(firehose.putRecordBatch);

    return firehose.putRecordBatchP(params).then(data => {
        if(data.FailedPutCount > 0){
            //collect the RequestResponses which are not processed.
            //index of those records is same as in request
            //process those failed records again
            console.log('Error',data);
        }
        return data;

    }).catch(error => {
        if(error.code == 'ServiceUnavailableException' && retry < retries.length){
            console.log('retryCount=',retry);

            // CHAIN the promise delay and retry here
            return delay(retries[retry]).then(() => {
                console.log('retry after delay #',retry);
                return batchWrite(records,firehose,retry+1); 
            });
        } else {
            // REJECT promise here upon error
            console.log('Error, rejecting promise');
            throw error;
        }

    });
}

有关您的代码的一些事情我不明白:

1. 当您从 putRecordBatch() 收到不可重试的错误时，您似乎应该拒绝该 promise ，以便调用者知道操作失败。我更改了代码来做到这一点。

2. 我不清楚 if(data.FailedPutCount > 0){...} 子句的用途，或者为什么您所做的只是其中的 console.log()。