使用 Vapor 3 创建和使用游标

Question

这可能是一堆蠕虫，我会尽力描述这个问题。我们有一个很长的运行ning 数据处理工作。我们的行动数据库被添加到每晚，并处理未完成的行动。处理夜间操作大约需要 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() 并且我收到错误 "Precondition failed: wait() must not be called when on the EventLoop"很公平。我面临的问题之一是，我不知道您是如何在后台线程上离开主事件循环以进行运行此类事情的。我知道 BlockingIOThreadPool，但它似乎仍然在同一个 EventLoop 上运行，并且仍然会导致错误。虽然我能够从理论上提出越来越复杂的方法来实现这一点，但我希望我错过了一个优雅的解决方案，也许对 SwiftNIO 和 Fluent 有更好了解的人可以提供帮助。

编辑：需要说明的是，这样做的目的显然不是汇总数据库中的操作数。目标是使用游标同步处理每个动作。当我读入结果时，我检测到动作的变化，然后将它们分批扔给处理线程。当所有线程都忙时，我不会再次从游标开始读取，直到它们完成。

有很多这样的动作，单个动作高达 4500 万次运行。聚合承诺和递归似乎不是一个好主意，当我尝试它时，服务器挂了。

这是一项处理密集型任务，可以在单个线程上运行数天，因此我不关心创建新线程。问题是我不知道如何在 Command 中使用 wait() 函数，因为我需要一个容器来创建数据库连接，而我唯一可以访问的是 context.container 在此调用 wait() 会导致上述错误。

TIA

Answer 1

好的，如您所知，问题出在以下几行：

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 return 每次传递一个整数，而 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 就是我们所得到的，我们 return 一个新的成功的未来，我们已经收集了迄今为止的一切

确实如此。我们这里做的是把同步循环变成异步递归。

好的，我们查看了很多代码，但这现在与您的功能有何关系？

信不信由你，但这确实有效（未经测试）：

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 直到它 returns nil 但不是累积所有 getPartial 的结果，它调用 process 获取部分结果并可以做一些进一步的处理。下一个 getPartial 调用将在 EventLoopFuture process return 完成时发生。

这更接近您想要的吗？

注意：我在这里使用了 SwiftNIO 的 EventLoopFuture 类型，在 Vapor 中你只需使用 Future 代替，但代码的其余部分应该是相同的。

Answer 2

这是通用解决方案，为 NIO 2.16/Vapor 4 重写，作为 EventLoop 的扩展

extension EventLoop {

    func accumulateResults<T>(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>(partialResultsSoFar: [T] /* our accumulator */,
                                  getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<[T]> {
            // let's run getPartial once
            return getPartial().flatMap { 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(partialResultsSoFar: partialResultsSoFar + [partialResult],
                                             getPartial: getPartial)
                } else {
                    // we've got all the partial results, yay, let's fulfill the overall future
                    return self.makeSucceededFuture(partialResultsSoFar)
                }
            }
        }
        return accumulateResults(partialResultsSoFar: [], getPartial: getPartial)
    }
}

使用 Vapor 3 创建和使用游标

Creating and consuming a cursor with Vapor 3

swift

vapor

swift-nio