如何在 ClojureScript / core.async 中停止 go 块？答案

【问题标题】：How to stop go block in ClojureScript / core.async?如何在 ClojureScript / core.async 中停止 go 块？
【发布时间】：2015-12-10 19:40:36
【问题描述】：

有没有优雅的方法来阻止正在运行的 go 块？

（不引入标志并使用检查/分支污染代码）

(ns example
  (:require-macros [cljs.core.async.macros :refer [go]])
  (:require        [cljs.core.async        :refer [<! timeout]]))

(defn some-long-task []
  (go
    (println "entering")

    ; some complex long-running task (e.g. fetching something via network)
    (<! (timeout 1000)) 
    (<! (timeout 1000))
    (<! (timeout 1000))
    (<! (timeout 1000))

    (println "leaving")))

; run the task
(def task (some-long-task))

; later, realize we no longer need the result and want to cancel it 
; (stop! task)

【问题讨论】：

(close! task) 也许？ - 谷歌第一个结果。

标签： clojure clojurescript core.async

【解决方案1】：

抱歉，这在今天的 core.async 中是不可能的。您从创建go 块中得到的结果是一个正常的通道，该块的结果将被放置在该通道上，尽管这并没有为您提供对实际块本身的任何处理。

【讨论】：

【解决方案2】：

您可能希望对此类任务使用 future 和 future-cancel。

(def f (future (while (not (Thread/interrupted)) (your-function ... ))))
(future-cancel f)

Why do cancelled Clojure futures continue using CPU?

【讨论】：

【解决方案3】：

正如 Arthur 的回答中所述，您不能立即终止 go 块，但由于您的示例指示多阶段任务（使用子任务），因此这样的方法可能会起作用：

(defn task-processor
  "Takes an initial state value and number of tasks (fns). Puts tasks
  on a work queue channel and then executes them in a go-loop, with
  each task passed the current state. A task's return value is used as
  input for next task. When all tasks are processed or queue has been
  closed, places current result/state onto a result channel. To allow
  nil values, result is wrapped in a map:

  {:value state :complete? true/false}

  This fn returns a map of {:queue queue-chan :result result-chan}"
  [init & tasks]
  (assert (pos? (count tasks)))
  (let [queue  (chan)
        result (chan)]
    (async/onto-chan queue tasks)
    (go-loop [state init, i 0]
      (if-let [task (<! queue)]
        (recur (task state) (inc i))
        (do (prn "task queue finished/terminated")
            (>! result {:value state :complete? (== i (count tasks))}))))
    {:queue  queue
     :result result}))

(defn dummy-task [x] (prn :task x) (Thread/sleep 1000) (inc x))

;; kick of tasks
(def proc (apply task-processor 0 (repeat 100 dummy-task)))

;; result handler
(go
  (let [res (<! (:result proc))]
    (prn :final-result res)))

;; to stop the queue after current task is complete
;; in this example it might take up to an additional second
;; for the terminated result to be delivered
(close! (:queue proc))

【讨论】：