error-handling - 使用 clojure core.async 管道处理错误

Question

我试图了解使用 core.async/pipeline 处理错误的正确方法是什么，我的管道如下:

input     --> xf-run-computation --> first-out
first-out --> xf-run-computation --> last-out

哪里 xf-run-computation将执行 http 调用并返回响应。但是，其中一些响应将返回错误。处理这些错误的最佳方法是什么？
我的解决方案是在 success-values 中拆分输出 channel 和 error-values然后将它们合并回一个 channel :

(let [[success-values1 error-values1] (split fn-to-split first-out)
      [success-values2 error-values2] (split fn-to-split last-out)
      errors (merge [error-values1 error-values2])]
(pipeline 4 first-out xf-run-computation input)
(pipeline 4 last-out  xf-run-computation success-values1)
[last-out errors])

所以我的函数将返回最后的结果和错误。

Answer 1

一般来说，什么是“正确的”方法可能取决于您的应用程序需求，但鉴于您的问题描述，我认为您需要考虑三件事:

xf-run-computation返回您的业务逻辑会视为错误的数据，

xf-run-computation抛出异常和

鉴于涉及 http 调用，一些运行 xf-run-computation可能永远不会完成(或无法及时完成)。

关于第 3 点，您应该考虑的第一件事是使用 pipeline-blocking而不是 pipeline .

我认为你的问题主要与第1点有关。基本思想是 xf-run-computation的结果。需要返回一个数据结构(比如 map 或记录)，它清楚地将结果标记为错误或成功，例如 {:title nil :body nil :status "error"} .这将为您提供一些处理这种情况的选择:

您后面的所有代码都忽略了具有 :status "error" 的输入数据.即，您的 xf-run-computation将包含类似 (when (not (= (:status input) "error")) (run-computation input)) 的行,

您可以对 pipeline 之间的所有结果运行过滤器-来电和 filter根据需要使用它们(注意 filter 也可以用作管道中的转换器，从而消除 core.async 旧的 filter> 和 filter< 功能)，

您使用 async/split就像你建议的那样/艾伦汤普森在他的回答中显示了将错误值过滤到一个单独的错误 channel 。如果您无论如何都要合并值，则没有真正需要为您的第二个管道设置第二个错误 channel ，您可以简单地重新使用您的错误 channel 。

对于第 2 点，问题在于 xf-run-computation 中的任何异常正在另一个线程中发生，不会简单地传播回您的调用代码。但是您可以使用 ex-handler论据 pipeline (和 pipeline-blocking )。您可以简单地过滤掉所有异常，将结果放在单独的异常 channel 中，或者 try catch 它们并将它们转换为错误(可能将它们放回结果或另一个错误 channel )——后者只有在以下情况下才有意义异常为您提供了足够的信息，例如一个 id 或允许将异常与导致异常的输入联系起来的东西。您可以在 xf-run-computation 中安排此事(即 catch 从第三方库抛出的任何异常，如 http 调用)。

对于第 3 点，core.async 中的规范答案是指向 timeout channel ，但这与 pipeline 没有多大意义.一个更好的主意是确保您的 http 调用设置了超时，例如 :timeout http-kit 或 :socket-timeout 选项和 :conn-timeout clj-http 的。请注意，这些选项通常会导致超时异常。