代码之家 › 专栏 › 技术社区 › hg.park

在promise.al()中下载s3客户端时丢失网络连接

promise amazon-s3 amazon-web-services node.js

0

hg.park · 技术社区 · 2 年前

我正在一个基于nodejs的服务器上工作,该服务器从s3下载1000个pdf文件,每个文件9MB,将它们压缩成zip文件,并返回URL。

但是,从S3下载时出现了问题。当我在S3上开始下载时,大约下载了100次,然后无法连接互联网。看起来它与大小无关,因为它在尝试使用1GB文件的5个文件时成功了。

这导致mysql连接丢失>事务失败 [ERROR] Error: BEGIN; - Can't add new command when connection is in closed state lost connection

我怀疑繁重的网络操作不应该放在promise.al()中,但我不知道确切的原因。帮助

await Promise.all(
  Ids.map((id) => {
    const readablePayloads: Array<{ filename: string; bugffer: Readable; }> = [];
    // get key, bucket by id
    
    const buffer = await S3Helper.getReadable({bucket, key} {accessKeyId, secretAccessKey});
    readablePayloads.push({ filename, buffer });
  })
)

static async getReadable(opts: { bucket: string; key: string }, awsOpts: AwsOptions): Promise<Readable> {
    return (await createS3Client(awsOpts).send(new AWS.GetObjectCommand({ Bucket: opts.bucket, Key: opts.key }))).Body as Readable;
  }

可以肯定的是,这段代码工作的连接不到100个

0 回复 | 直到 2 年前

1

0

Edgar P-Yan 2 年前

我认为造成这个问题的可能原因很少:

文件描述符限制。
任何打开的连接都使用文件描述符。Linux的默认限制为 file descriptors per process is 1024 。因此,除了分叉node.js进程或增加限制之外,您对此无能为力。除非你有充分的理由,否则不建议使用后者。另一种方法是限制连接的数量,我将在下面介绍。
内存堆限制。
这种可能性较小,但仍然是:每个打开的连接都会使用一定量的内存。此外,根据您编写代码和处理可读流的方式,您可能会在可读流中有过多的缓冲,甚至内存泄漏。

限制连接数

以较小的组下载文件,而不是一次全部下载。它可能看起来像这样:

// _.chunk comes from the `lodash` library.
// Now `chunksOfIds` is an array of arrays, each of which contains up to 100 ids.
// i.e [[1,2,...,100],[101,102,...,200],...,[1001,1002,...,1100]]
const chunksOfIds = _.chunk(Ids, 100)

for (const chunk of chunksOfIds) {
  const readablePayloadsOfThisChunk: Array<{ filename: string; bugffer: Readable; }> = [];

  chunk.map((id) => {
    // get key, bucket by id

    const buffer = await S3Helper.getReadable({bucket, key} {accessKeyId, secretAccessKey});
    readablePayloadsOfThisChunk.push({ filename, buffer });
  })


  // Note: you have to process your `readablePayloads` here, in the loop,
  // so that you will finish your computations before the processing of the next chunk will start
  await processReadablePayloads(readablePayloadsOfThisChunk)
}