Clojure 中的死简单 Fork-Join 并发答案

【问题标题】：Dead simple Fork-Join concurrency in ClojureClojure 中的死简单 Fork-Join 并发
【发布时间】：2012-09-07 01:48:39
【问题描述】：

我有两个独立的昂贵功能。我想并行运行它们。我不想处理期货之类的事情（我是 Clojure 的新手，很容易混淆）。

我正在寻找一种同时运行两个函数的简单方法。我希望它像下面这样工作

(defn fn1 [input] ...) ; costly
(defn fn2 [input] ...) ; costly

(let [[out1 out2] (conc (fn1 x) (fn2 y))] ...)

我希望它返回一个带有一对输出的向量。它应该只在两个线程都终止后才返回。理想情况下， conc 应该适用于任意数量的输入。我怀疑这是一个简单的模式。

【问题讨论】：

当您说您不想处理期货时，是否意味着您也不想在“conc”函数中使用期货？据我所知，在这种情况下使用 Clojure 并发原语之一是惯用的，尽管它们可以通过封装在“conc”中对您隐藏。
肯定会用到一些并发原语。 conc 可以随心所欲。我只是不想以用户的身份与他们打交道。我怀疑这是“为每个输入开始一个未来”，“等待每个输出”，“返回”。也许它必须是一个宏，不确定。
如果您想将参数的评估推迟到线程，那么绝对是一个宏。我现在正在处理宏定义。

标签： concurrency clojure future

【解决方案1】：

在 Clojure 中使用期货非常容易。无论如何，这是一个避免它们的答案

(defn conc [& fns]
  (doall (pmap (fn [f] (f)) fns)))

pmap 在后台使用期货。 doall 将强制序列进行评估。

(let [[out1 out2] (conc fn1 fn2)]
        [out1 out2])

请注意，我解构了 out1 和 out2 以试图保留您的示例。

【讨论】：

要使用它，您需要将输入包装在匿名函数中，即：#(fn1 42)
@ArthurUlfeldt 正确，在这种情况下，我会选择 JohnJ 的解决方案。
我认为不将其包装在宏中会导致更好的可组合代码。宏不是一流的东西，您不能将它们传递给映射或将它们应用于参数列表。

【解决方案2】：

您确实需要一个宏来保留所需的语法，尽管如其他答案所示，还有其他方法可以获得相同的行为。这是一种方法：

(defn f1 [x] (Thread/sleep 500) 5)
(defn f2 [y] 2)

(defmacro conc [& exprs]
  `(map deref
        [~@(for [x# exprs] `(future ~x#))]))

(time (let [[a b] (conc (f1 6) (f2 7))]
       [a b]))
; "Elapsed time: 500.951 msecs"
;= (5 2)

扩展展示了它是如何工作的：

(macroexpand-1 '(conc (f1 6) (f2 7)))
;= (clojure.core/map clojure.core/deref [(clojure.core/future (f1 6)) 
;=                                       (clojure.core/future (f2 7))])

您指定了两个函数，但这应该适用于任意数量的表达式。

【讨论】：

太棒了。这激发了我自己尝试的灵感。我试图通过将 for 替换为 (map future exprs) 来实现它。可悲的是，这不起作用，因为未来本身就是一个宏。关于如何进一步减少这种情况的任何想法？
是的，这就是宏的问题——不能像函数那样简单地组合它们。我不知道如何进一步减少它，但Google Group 上的某个人可能会。

【解决方案3】：

我知道您不希望您的最终解决方案公开期货，尽管说明如何使用期货做到这一点很有用，然后将它们包装在隐藏此细节的东西中：

core> (defn fn1 [input] (java.lang.Thread/sleep 2000) (inc input))
#'core/fn1                                                                                     
core> (defn fn2 [input] (java.lang.Thread/sleep 3000) (* 2 input))
#'core/fn2                                                                                     
core> (time (let [f1 (future (fn1 4)) f2 (future (fn2 4))] @f1 @f2))
"Elapsed time: 3000.791021 msecs"

然后我们可以将它包装在围绕期货的许多 clojure 包装器中。最简单的就是一个函数，它接受两个函数并并行运行它们。

core> (defn conc [fn1 fn2] 
         (let [f1 (future (fn1)) 
               f2 (future (fn2))] [@f1 @f2]))
#'core/conc                                                                                    
core> (time (conc #(fn1 4) #(fn2 4)))
"Elapsed time: 3001.197634 msecs"

这避免了将其编写为宏的需要，方法是让 conc 运行函数而不是评估主体，然后通过将 # 放在调用前面来创建要传递给它的函数。

这也可以用map和future-call来写：

core> (map deref (map future-call [#(fn1 4) #(fn2 42)]))
(5 84)

然后你可以提高 conc 直到它类似于（正如 Julien Chastang 明智地指出的那样）pmap

【讨论】：