【发布时间】:2013-08-19 13:28:57
【问题描述】:
我有许多(未计算的)表达式保存在一个向量中; [ expr1 expr2 expr3 ... ]
我想做的是将每个表达式交给一个单独的线程并等待一个返回值。那时我对其他线程的结果不感兴趣,想取消它们以节省 CPU 资源。
(我意识到这可能会导致不确定性,因为程序的不同运行可能会导致首先评估不同的表达式。我手头有这个。)
是否有实现上述目标的标准/惯用方式?
【问题讨论】:
标签: concurrency clojure
我有许多(未计算的)表达式保存在一个向量中; [ expr1 expr2 expr3 ... ]
我想做的是将每个表达式交给一个单独的线程并等待一个返回值。那时我对其他线程的结果不感兴趣,想取消它们以节省 CPU 资源。
(我意识到这可能会导致不确定性,因为程序的不同运行可能会导致首先评估不同的表达式。我手头有这个。)
是否有实现上述目标的标准/惯用方式?
【问题讨论】:
标签: concurrency clojure
这是我的看法。
基本上你必须在你的每个future中解析一个全局promise,然后返回一个包含future列表和resolved值的向量,然后取消列表中的所有future:
(defn run-and-cancel [& expr]
(let [p (promise)
run-futures (fn [& expr] [(doall (map #(future (deliver p (eval %1))) expr)) @p])
[fs res] (apply run-futures expr)]
(map future-cancel fs)
res))
【讨论】:
它尚未正式发布,但core.async 看起来可能是一种有趣的方式来解决您的问题 - 以及其他异步问题,非常巧妙。
core.async 的 leiningen 咒语(目前)如下:
[org.clojure/core.async "0.1.0-SNAPSHOT"]
这里有一些代码可以创建一个函数,该函数需要一些耗时的函数,并阻塞直到其中一个返回。
(require '[clojure.core.async :refer [>!! chan alts!! thread]]))
(defn return-first [& ops]
(let [v (map vector ops (repeatedly chan))]
(doseq [[op c] v]
(thread (>!! c (op))))
(let [[value channel] (alts!! (map second v))]
value)))
;; Make sure the function returns what we expect with a simple Thread/sleep
(assert (= (return-first (fn [] (Thread/sleep 3000) 3000)
(fn [] (Thread/sleep 2000) 2000)
(fn [] (Thread/sleep 5000) 5000))
2000))
在上面的示例中:
chan 创建异步通道>!! 将值放入通道thread 在另一个线程中执行主体alts!! 采用通道向量,并在其中任何一个出现值时返回还有比这更多的方法,我仍在摸索,但这里有一个演练:https://github.com/clojure/core.async/blob/master/examples/walkthrough.clj
David Nolen 的博客上有一些很棒的帖子,如果令人难以置信的话 (http://swannodette.github.io/)
编辑
刚刚看到 Michał Marczyk 回答了一个非常相似的问题,但更好的是,在这里,并且它允许您取消/短路。 with Clojure threading long running processes and comparing their returns
【讨论】:
你想要的是 Java 的CompletionService。我不知道clojure中有任何包装器,但是使用互操作并不难。下面的示例大致基于ExecutorCompletionService 的JavaDoc 页面上的示例。
(defn f [col]
(let [cs (ExecutorCompletionService. (Executors/newCachedThreadPool))
futures (map #(.submit cs %) col)
result (.get (.take cs))]
(map #(.cancel % true) futures)
result))
【讨论】:
ExecutorCompletionService 是一个很好的建议。另一方面,您的代码将无法工作,因为它忽略了 map 的懒惰:没有任何任务将被提交,.take 调用将永远阻塞。
map的意见;我没有注意到这一点。我想我可以将地图调用包装在doall?
doall 中。对于.cancel 调用,使用doseq 会更加惯用。
您可以使用 future-call 来获取所有期货的列表,并将它们存储在 Atom 中。然后,用“在头部射击其他未来”功能组合每个正在运行的未来,这样第一个将终止所有剩余的未来。 Here is an example:
(defn first-out [& fns]
(let [fs (atom [])
terminate (fn [] (println "cancling..") (doall (map future-cancel @fs)))]
(reset! fs (doall (map (fn [x] (future-call #((x) (terminate)))) fns)))))
(defn wait-for [n s]
(fn [] (print "start...") (flush) (Thread/sleep n) (print s) (flush)))
(first-out (wait-for 1000 "long") (wait-for 500 "short"))
编辑
刚刚注意到前面的代码没有返回第一个结果,所以它主要用于副作用。这是another version,它使用承诺返回第一个结果:
(defn first-out [& fns]
(let [fs (atom [])
ret (promise)
terminate (fn [x] (println "cancling.." )
(doall (map future-cancel @fs))
(deliver ret x))]
(reset! fs (doall (map (fn [x] (future-call #(terminate (x)))) fns)))
@ret))
(defn wait-for [n s]
"this time, return the value"
(fn [] (print "start...") (flush) (Thread/sleep n) (print s) (flush) s))
(first-out (wait-for 1000 "long") (wait-for 500 "short"))
【讨论】:
虽然我不知道是否有一种惯用的方式来实现您的目标,但 Clojure Future 看起来很合适。
接受一个表达式体并产生一个未来对象,它将 在另一个线程中调用主体,并将缓存结果并 在所有后续调用 deref/@ 时返回它。如果计算有 尚未完成,对 deref/@ 的调用将阻塞,除非 使用带超时的 deref。
【讨论】: