Groovy PromiseMap - 我可以限制异步线程池吗？答案

【问题标题】：Groovy PromiseMap - Can I limit the asynchronous thread pool?Groovy PromiseMap - 我可以限制异步线程池吗？
【发布时间】：2014-06-16 18:43:05
【问题描述】：

我正在制作一个（快速而肮脏的）批处理 API，它允许 UI 发送选择的 REST API 调用并立即为所有这些调用获取结果。

我正在使用 PromiseMap 对相关服务进行一些异步 REST 调用，这些服务随后会被收集。

可能有大量线程需要运行，我想限制同时运行的线程数，类似于Executor的线程池。

如果不将线程物理分离成多个 PromiseMap 并将它们链接起来，这是否可行？我没有在网上找到任何描述限制线程池的内容。

//get requested calls
JSONArray callsToMake=request.JSON as JSONArray 

//registers calls in promise map
def promiseMap = new PromiseMap()
//Can I limit this Map as a thread pool to, say, run 10 at a time until finished

data.each {
def tempVar=it
promiseMap[tempVar.id]={makeCall(tempVar.method, "${basePath}${tempVar.to}" as String, tempVar.body)}
}

def result=promiseMap.get()
def resultList=parseResults(result)
response.status=HttpStatusCodes.ACCEPTED
render resultList as JSON

我希望有一个我可能不知道的相当直接的设置。

谢谢。

【问题讨论】：

protip，你不需要线程来进行多个异步调用，只要不阻塞就可以了
Benjamin，其意图是阻止此调用的返回，直到所有这些异步调用完成并收集它们的结果并立即将它们发送回来。阻塞是有意的，线程是专门允许阻塞和收集响应的——如果有另一种方法可以使一堆异步然后阻塞直到全部回来，我愿意接受你可能提供的任何建议。我的问题是关于如何节流，以便 1000 个调用不会立即破坏服务器。你会提出什么来实现这些目标？
我不知道 :D 公平地说，在 Scala（设置 ExecutionContext）、JavaScript（{concurrency: 8}）或 C#（.WithDegreeOfParallelism(8)）中，我知道答案。你当然可以自己实现。
当然，我知道如何使用其他不太理想的方法来做到这一点。 Promise 是 Groovy 的多线程，所以我希望已经内置了一种方法来限制线程。

标签： grails asynchronous groovy promise

【解决方案1】：

Grails 中的默认异步实现是 GPars。要配置线程数，您需要使用 GParsPool。见：

http://gpars.org/guide/guide/dataParallelism.html#dataParallelism_parallelCollections_GParsPool

例子：

withPool(10) {...}

【讨论】：

嗨格雷姆。好的，所以从它的声音来看，GPars 会比使用 Promises 更好。我认为 Promises 是较新版本中的默认设置，因为它们位于 grails.async 包中？无论如何，我必须更好地阅读您的链接，但这听起来像是我们想要的。 GPars 可以阻止控制器返回，直到所有线程都完成？
我今天无法尝试这个，但文档看起来就像我需要的那样，所以我要感谢你，Graeme。谢谢！
在 grails-async 3 (4?) 中使用 GPars 仍然是实现可重用线程池的推荐方法，还是有一些内置机制？

【解决方案2】：

withPool 似乎不起作用。以防万一有人想在这里限制线程，这就是我所做的。我们可以使用自定义的 ThreadPool 创建一个自定义 Group 并指定线程的数量。

def customGroup = new DefaultPGroup(new DefaultPool(true, 5))
try {
  Dataflow.usingGroup(customGroup, {
    def promises = new PromiseList()
    (1..100).each { number ->
      promises << {
        log.info "Performing Task ${number}"
        Thread.sleep(200)
        number++
      }
    }
    def result = promises.get()
  })
}
finally {
  customGroup.shutdown()
}

【讨论】：