请求承诺导致打开太多套接字

Question

我目前正在尝试编写一个函数，该函数接受用户输入和 returns 根据训练数据中单词出现的次数，用户评论可能属于的顶级 subreddit。我有一个包含不同 subreddit 中单词频率的数据库，我正在为每个 subreddit 和用户输入中的每个单词发出 GET 请求。

这加起来有很多 get 请求，因为我的数据库中有超过 7000 个 subreddits。我目前正在发出请求承诺（浅层）请求以获取所有 subreddits 的列表。然后，对于每个 subreddit，我循环遍历用户输入中的每个单词，创建另一个请求承诺对象并将其添加到承诺数组中。

一旦添加了所有请求-承诺对象，我就等到它们都使用 Promise.all 解决，然后我尝试打印出给定 subreddit 的词频数组，但我得到了一个'Error: connect EMFILE' 消息。

根据另一个 post 关于堆栈溢出的说法，这意味着我打开了太多的套接字，但我不明白这是怎么发生的。根据我的理解，它不会一次最多打开 user_words.length 个可能的连接，因为这些是在 Promise.all 等待解决时正在完成的请求？我看不出连接是如何关闭的。

在此先感谢您的帮助！

function getBestSubreddit(messageText) {
  var user_words = parse_message(messageText);
  var top_subreddit = "";
  var top_score = Number.MIN_SAFE_INTEGER;
  rp(dbUrl + '/.json?shallow=true').then(function(res) {
    res = JSON.parse(res);
    for (var subreddit in res) {
      if (res.hasOwnProperty(subreddit)) {
        var score = 0.0;
        var promises = []
        for (var i = 0; i < user_words.length; i++) {
          promises.push(rp(dbUrl + '/' + subreddit + '/word_freqs/' + user_words[i] + '.json'));
        }
        Promise.all(promises).then(function(values) {
          console.log(values);
        }, function(err) {
          console.log(err);
        });
      }
    }
  }).catch(function(err) {
    console.log(err);
  })
  return top_subreddit;
}

Answer 1

From my understanding, wouldn't it only ever have up to user_words.length possible connections open at a time, since those are the requests that are being done while the Promise.all waits to resolve? I don't see how the connections are not being closed.

不，这是不正确的。您有两个嵌套的 for 循环，因此您可以同时打开多达 user_words.length * how many subreddits there are 个循环。请记住，rp() 和 Promise.all() 不会阻塞，因此您运行嵌套 for 循环完成，在处理任何响应之前启动每个连接。

看起来您还希望以某种方式同步 return 代码行 return top_subreddit 的结果。你也不能那样做。您应该 return 一个最终会解决所需结果的承诺。

From my understanding, wouldn't it only ever have up to user_words.length possible connections open at a time, since those are the requests that are being done while the Promise.all waits to resolve? I don't see how the connections are not being closed.

这不是Promise.all()的正确理解。 Promise.all() 没有阻塞。在您的代码继续退出之前，直到所有承诺都已解决，它才会 "wait" 。它的行为是异步的。您的代码继续执行 for 循环的其他迭代，并且 Promise.all() 在您传递给它的所有承诺都已完成时的某个时间调用它的 .then() 处理程序。 for 循环的其他迭代继续运行并堆积更多套接字。

我认为解决这个问题的最简单方法是创建一个您要处理的 URL 数组，然后使用一个已经内置函数的异步库来允许您运行同时进行最多 N 个异步操作。由于您的代码是基于承诺的，我会选择 Bluebird 的 Promise.map() 来处理 URL 列表。

var Promise = require('bluebird');

function getBestSubreddit(messageText) {
  var user_words = parse_message(messageText);
  var top_subreddit = "";
  var top_score = Number.MIN_SAFE_INTEGER;
  return rp(dbUrl + '/.json?shallow=true').then(function(res) {
    res = JSON.parse(res);
    // build a list of URLs to process
    var urls = [];
    for (var subreddit in res) {
      if (res.hasOwnProperty(subreddit)) {
        for (var i = 0; i < user_words.length; i++) {
          urls.push(dbUrl + '/' + subreddit + '/word_freqs/' + user_words[i] + '.json');
        }
      }
    }
    // 
    return Promise.map(urls, function(url) {
        return rp(url);
    }, {concurrency: 20}).then(function(allResults) {
        // do any final processing of allResults here and return that value
        // to become the resolved result of the returned promise
    });
  }
}

getBestSubreddit(someText).then(function(result) {
    // process result here
}).catch(function(err) {
    // handle error here
});

本例我设置的并发数为20。您可以试验将其更改为更高或更低的数字是否会提高您的吞吐量。理想的数量取决于许多因素，包括您的本地执行环境、您请求的数据量和您拥有的带宽以及您发出请求的目标主机以及它如何处理同时请求。如果您太快发出太多请求，您可能还需要担心目标的速率限制。

其他一些相关回答：

从你的问题中我仍然不清楚你想要得到什么结果，但这里有一个收集所有可能数据的版本。您最终会得到这种形式的对象数组 {result: result, subreddit: subreddit, word: word}，其中 result 是您对给定的 subreddit 和给定单词的 rp() 的结果。然后，您可以根据需要整理该组结果：

var Promise = require('bluebird');

function getBestSubreddit(messageText) {
  var user_words = parse_message(messageText);
  var top_subreddit = "";
  var top_score = Number.MIN_SAFE_INTEGER;
  return rp(dbUrl + '/.json?shallow=true').then(function(res) {
    res = JSON.parse(res);
    // build a list of URLs to process
    var requestData = [];
    for (var subreddit in res) {
      if (res.hasOwnProperty(subreddit)) {
        for (var i = 0; i < user_words.length; i++) {
          requestData.push({url:dbUrl + '/' + subreddit + '/word_freqs/' + user_words[i] + '.json', subreddit: subreddit, word: user_words[i]});
        }
      }
    }
    // 
    return Promise.map(requestData, function(url) {
        return rp(requestData.url).then(function(result) {
            return {result: result, subreddit: requestData.subreddit, word: requestData.word};
        });
    }, {concurrency: 20}).then(function(allResults) {
        // now filter through all the data with appropriate subreddit
        // allResults is an array of objects of this form {result: result, subreddit: subreddit, word: word}
        // return whatever you want the final result to be after processing the allResults array
    });
  }
}

getBestSubreddit(someText).then(function(result) {
    // process result here
}).catch(function(err) {
    // handle error here
});

Answer 2

问题源于两个嵌套循环和不受限制的 rp() 调用，导致许多同时请求。

节流通常通过以下方式实现：

通过构建 then() 链进行序列化，例如通过减少数组。
强制执行 "concurrency" 限制，例如使用 Bluebird 的 Promise.map() 及其 concurrency 选项。

我想一定有很多解决这个特定问题的方法，但本质上是：

集中所有请求并通过并发限制（jFriend00 的回答），
允许一个循环保持同步并通过序列化或并发限制另一个循环，
在序列化中嵌套序列化，
在一个并发中嵌套一个并发，
采用序列化和并发的混合方法。

这是一种混合方法，其中：

原来的外循环被序列化限制了
原始内循环被 Bluebird 的并发映射限制了。

function getSubreddits(messageText) {
    var user_words = parse_message(messageText);
    return rp(dbUrl + '/.json?shallow=true').then(function(res) {
        var subreddits = Object.keys(JSON.parse(res));
        return subreddits.reduce(function(p, subreddit) {
            return p.then(function() {
                return Promise.map(user_words, function(word) {
                    return rp(dbUrl + '/' + subreddit + '/word_freqs/' + word + '.json');
                }, {concurrency: 10}).then(function(freqs) {
                    // return an object that associates each subreddit with its results
                    return {
                        'subreddit': subreddit, // or maybe the object for which `subreddit` is the key?
                        'word_freqs': freqs
                    };
                });
            });
        }, Promise.resolve());
    });
}

一个缺点是您最终会遇到嵌套很深的问题，这不适合展平。也就是说，大多数（如果不是全部）其他方法都是相似的。

无论您采用哪种方法，getBestSubreddit() 现在都将包含对 getSubreddits() 的调用以及对结果的一些 post 处理。

function getBestSubreddit(messageText) {
    return getSubreddits(messageText).then(function(results) {
        // Here `results` is an array of `{'subreddit', 'word_freqs'}` objects.
        // Loop through and calculate a score for each subreddit,
        // then use that score to determine the top subreddit,
        // and return it.
    }).catch(function(error) {
        console.log(error);
    });
}

请求承诺导致打开太多套接字

request-promise causing too many sockets to be open

request

node.js

promise

request-promise