好的,如您所知,问题出在以下几行:
while ... {
...
try connection.raw("...").all(decoding: RecRegAction.self).wait()
...
}
您想等待许多结果,因此您使用while
循环和.wait()
所有中间结果。本质上,这是在事件循环上将异步代码转换为同步代码。这很可能会导致死锁,并且肯定会停止其他连接,这就是为什么 SwiftNIO 会尝试检测并给你那个错误的原因。我不会详细说明为什么它会停止其他连接,或者为什么这可能会导致这个答案出现死锁。
让我们看看我们有哪些选项可以解决这个问题:
- 正如您所说,我们可以将它
.wait()
放在另一个不是事件循环线程之一的线程上。为此,任何非EventLoop
线程都可以:要么 aDispatchQueue
要么您可以使用BlockingIOThreadPool
( 它不在 a 上运行EventLoop
)
- 我们可以将您的代码重写为异步的
两种解决方案都可以使用,但 (1) 确实不可取,因为您会烧掉整个(内核)线程来等待结果。并且两者Dispatch
都有BlockingIOThreadPool
他们愿意产生的有限数量的线程,所以如果你经常这样做,你可能会用完线程,所以需要更长的时间。
因此,让我们看看如何在累积中间结果的同时多次调用异步函数。然后,如果我们累积了所有中间结果,则继续所有结果。
为了使事情更容易,让我们看一个与您的非常相似的函数。我们假设这个函数就像在你的代码中一样提供
/// delivers partial results (integers) and `nil` if no further elements are available
func deliverPartialResult() -> EventLoopFuture<Int?> {
...
}
我们现在想要的是一个新功能
func deliverFullResult() -> EventLoopFuture<[Int]>
请注意deliverPartialResult
每次返回一个整数并deliverFullResult
传递一个整数数组(即所有整数)。好的,那么我们如何在deliverFullResult
不调用的情况下编写代码deliverPartialResult().wait()
?
那这个呢:
func accumulateResults(eventLoop: EventLoop,
partialResultsSoFar: [Int],
getPartial: @escaping () -> EventLoopFuture<Int?>) -> EventLoopFuture<[Int]> {
// let's run getPartial once
return getPartial().then { partialResult in
// we got a partial result, let's check what it is
if let partialResult = partialResult {
// another intermediate results, let's accumulate and call getPartial again
return accumulateResults(eventLoop: eventLoop,
partialResultsSoFar: partialResultsSoFar + [partialResult],
getPartial: getPartial)
} else {
// we've got all the partial results, yay, let's fulfill the overall future
return eventLoop.newSucceededFuture(result: partialResultsSoFar)
}
}
}
鉴于accumulateResults
,实施deliverFullResult
不再太难:
func deliverFullResult() -> EventLoopFuture<[Int]> {
return accumulateResults(eventLoop: myCurrentEventLoop,
partialResultsSoFar: [],
getPartial: deliverPartialResult)
}
但让我们更深入地了解accumulateResults
它的作用:
- 它调用
getPartial
一次,然后当它回调它
- 检查我们是否有
- 一个部分结果,在这种情况下,我们将它与另一个一起记住
partialResultsSoFar
并返回(1)
nil
这意味着partialResultsSoFar
我们所得到的一切,我们用迄今为止收集的一切返回一个新的成功的未来
这已经是真的了。我们在这里所做的是将同步循环变成异步递归。
好的,我们查看了很多代码,但现在这与您的函数有什么关系?
信不信由你,但这实际上应该有效(未经测试):
accumulateResults(eventLoop: el, partialResultsSoFar: []) {
connection.raw("FETCH \(chunkSize) FROM \(cursorName)")
.all(decoding: RecRegAction.self)
.map { results -> Int? in
if results.count > 0 {
return results.count
} else {
return nil
}
}
}.map { allResults in
return allResults.reduce(0, +)
}
所有这一切的结果将是一个EventLoopFuture<Int>
带有所有中间值的总和result.count
。
当然,我们首先将您所有的计数收集到一个数组中,然后allResults.reduce(0, +)
在最后将其相加(),这有点浪费,但也不是世界末日。我这样保留它是因为它accumulateResults
可以在您想要在数组中累积部分结果的其他情况下使用。
现在最后一件事,一个真正的accumulateResults
函数可能对元素类型是通用的,我们也可以消除partialResultsSoFar
外部函数的参数。那这个呢?
func accumulateResults<T>(eventLoop: EventLoop,
getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<[T]> {
// this is an inner function just to hide it from the outside which carries the accumulator
func accumulateResults<T>(eventLoop: EventLoop,
partialResultsSoFar: [T] /* our accumulator */,
getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<[T]> {
// let's run getPartial once
return getPartial().then { partialResult in
// we got a partial result, let's check what it is
if let partialResult = partialResult {
// another intermediate results, let's accumulate and call getPartial again
return accumulateResults(eventLoop: eventLoop,
partialResultsSoFar: partialResultsSoFar + [partialResult],
getPartial: getPartial)
} else {
// we've got all the partial results, yay, let's fulfill the overall future
return eventLoop.newSucceededFuture(result: partialResultsSoFar)
}
}
}
return accumulateResults(eventLoop: eventLoop, partialResultsSoFar: [], getPartial: getPartial)
}
编辑:编辑后,您的问题表明您实际上并不想累积中间结果。所以我的猜测是,您希望在收到每个中间结果后进行一些处理。如果这就是你想要做的,也许试试这个:
func processPartialResults<T, V>(eventLoop: EventLoop,
process: @escaping (T) -> EventLoopFuture<V>,
getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<V?> {
func processPartialResults<T, V>(eventLoop: EventLoop,
soFar: V?,
process: @escaping (T) -> EventLoopFuture<V>,
getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<V?> {
// let's run getPartial once
return getPartial().then { partialResult in
// we got a partial result, let's check what it is
if let partialResult = partialResult {
// another intermediate results, let's call the process function and move on
return process(partialResult).then { v in
return processPartialResults(eventLoop: eventLoop, soFar: v, process: process, getPartial: getPartial)
}
} else {
// we've got all the partial results, yay, let's fulfill the overall future
return eventLoop.newSucceededFuture(result: soFar)
}
}
}
return processPartialResults(eventLoop: eventLoop, soFar: nil, process: process, getPartial: getPartial)
}
这将(和以前一样)运行getPartial
直到它返回nil
,但它不会累积所有getPartial
的结果,而是调用process
它来获取部分结果并可以做一些进一步的处理。下一次getPartial
调用将在EventLoopFuture
process
返回完成时发生。
这更接近你想要的吗?
注意:我在这里使用了 SwiftNIO 的EventLoopFuture
类型,在 Vapor 中你只需要使用Future
它,但其余代码应该是相同的。