这可能是一 jar 蠕虫,我会尽力描述这个问题。我们有一个长期运行的数据处理工作。我们的行动数据库会每晚添加一次,并且会处理未完成的行动。处理每晚的操作大约需要15分钟。在Vapor 2中,我们利用了很多原始查询来创建PostgreSQL游标并循环遍历它,直到它为空。
目前,我们通过命令行参数运行该处理。将来,我们希望它作为主服务器的一部分运行,以便在执行处理时可以检查进度。
func run(using context: CommandContext) throws -> Future<Void> {
let table = "\"RecRegAction\""
let cursorName = "\"action_cursor\""
let chunkSize = 10_000
return context.container.withNewConnection(to: .psql) { connection in
return PostgreSQLDatabase.transactionExecute({ connection -> Future<Int> in
return connection.simpleQuery("DECLARE \(cursorName) CURSOR FOR SELECT * FROM \(table)").map { result in
var totalResults = 0
var finished : Bool = false
while !finished {
let results = try connection.raw("FETCH \(chunkSize) FROM \(cursorName)").all(decoding: RecRegAction.self).wait()
if results.count > 0 {
totalResults += results.count
print(totalResults)
// Obviously we do our processing here
}
else {
finished = true
}
}
return totalResults
}
}, on: connection)
}.transform(to: ())
}
现在这不起作用,因为我正在调用 wait(),并且收到错误“前提条件失败:在EventLoop上不得调用wait()” ,这很公平。我面临的问题之一是,我不知道您如何离开主事件循环,以在后台线程上运行类似的事情。我知道BlockingIOThreadPool,但这似乎仍然可以在同一EventLoop上运行,并且仍然会导致错误。虽然我可以从理论上提出越来越多的复杂方法来实现这一目标,但我希望我缺少一个优雅的解决方案,也许对SwiftNIO和Fluent有更深入了解的人可以提供帮助。
编辑:明确地说,这样做的目的显然是不总计数据库中的操作数。目标是使用游标同步处理每个 Action 。当我读入结果时,我检测到 Action 中的更改,然后将它们的批处理丢给处理线程。当所有线程都忙时,直到它们完成,我才从游标再次开始读取。
这些 Action 很多,单次运行最多4500万次。聚集promise和递归似乎不是一个好主意,当我尝试使用它时,仅出于此目的,服务器挂起了。
这是一项处理密集型任务,可以在单个线程上运行几天,因此我不必担心创建新线程。问题是我无法弄清楚如何在命令中使用wait()函数,因为我需要一个容器来创建数据库连接,而我只能访问的一个容器就是 context.container 调用wait()这样会导致上述错误。
TIA
最佳答案
好的,正如您所知道的,问题出在以下几行:
while ... {
...
try connection.raw("...").all(decoding: RecRegAction.self).wait()
...
}
您想等待许多结果,因此对所有中间结果都使用
while
循环和.wait()
。本质上,这是在事件循环上将异步代码转换为同步代码。这很可能导致死锁,并且肯定会导致其他连接停滞,这就是SwiftNIO尝试检测到该错误并为您提供错误的原因。我不会详细说明为什么它会拖延其他连接,或者为什么这很可能导致此答案陷入僵局。让我们看看我们必须解决什么问题:
就像您说的
.wait()
放在不是事件循环线程之一的另一个线程上。为此,任何非EventLoop
线程都可以:DispatchQueue
或您可以使用BlockingIOThreadPool
(不在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
就是我们所能得到的,我们将返回迄今为止已收集的所有内容的新成功的 future 确实是这样。我们在这里所做的是将同步循环变成异步递归。
好的,我们看了很多代码,但是现在这与您的功能有什么关系?
信不信由你,但这应该可以正常工作(未经测试):
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
,但是其余代码应该相同。
关于swift - 使用Vapor 3创建和使用游标,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51625050/