你如何杀死一个core.async/thread？答案

【问题标题】：How do you kill a core.async/thread?你如何杀死一个core.async/thread？
【发布时间】：2016-06-02 17:07:50
【问题描述】：

core.async/thread 返回的一些线程涉及某个进程，我将要关闭这些线程。我没有关闭我的整个程序，只是这些线程。如何终止线程？

Java Thread 类的.stop 方法已被弃用，但我很乐意使用它，除了core.async/thread 返回的不是Thread，而是ManyToManyChannel：

user=> (clojure.core.async/thread)
#object[clojure.core.async.impl.channels.ManyToManyChannel 0x780e97c0
  "clojure.core.async.impl.channels.ManyToManyChannel@780e97c0"]
user=> (type *1)
clojure.core.async.impl.channels.ManyToManyChannel

我没有在ManyToManyChannel 上找到任何文档。对于线程类型，这听起来像是一个奇怪的名称，所以这里可能有一些我不理解的基本内容。但这是我目前的幼稚、荒谬的问题：你如何杀死 ManyToManyChannel？

clojure.repl/thread-stopper 似乎对ManyToManyChannels 没有影响。

【问题讨论】：

标签： multithreading clojure terminate core.async

【解决方案1】：

你让线程自然终止。如果需要外部终止，则必须实现它。

(defn terminatable [input-ch terminate-ch]
   (thread
     (loop []
       (let [[v ch] (alts!! [input-ch terminate-ch])]
         (if (identical? ch input-ch)
           (if (some? v)
             (do (process-input v) (recur))
             ;; else input-ch has closed -> don't call recur,
             ;; thread terminates
             )
           ;; else we received sth. from terminate-ch,
           ;; or terminate-ch has closed -> don't call recur,
           ;; thread terminates
           )))))

然后通过外部终止 (close! terminate-ch)

最后你可以从thread返回的通道中判断线程何时终止。

我。 e.

(take! (terminatable (chan) (doto (chan) close!)) 
       (fn [_] (println "Thread is terminated")))

【讨论】：

这是否意味着我不能写这样的代码？ (async/thread (doseq [line (line-seq subprocess-stdout)] (>!! stdout-chan line))（也就是说，每行调用一次line-seq 而不是alts!!。）如果没有，是否有某种方法可以方便地调用line-seq 从子进程中读取所有内容，并且仍然有无需等待子进程退出即可轻松关闭它的方法？
这是流处理的重点：您使用通道对流进行建模。所以你必须把这些行放在一个通道上，就像我的例子中的 input-ch 一样，然后一次取一行。 IE。 (onto-chan input-ch (line-seq subprocess-stdout))。然后v变成一行，你可以做进一步的处理，比如在process-input的实现中把它放到stdout-chan上。
考虑投资pipeline-blocking 和pipeline-async 以及传感器，因为它们提供了简单的工具来并行处理您的流。 pipeline-blocking 实现了另一种终止策略（关闭目标通道）。
如果您只想将这些行放到 stdout-chan 上，您也可以使用(onto-chan stdout-chan (line-seq subprocess-stdout)) - 这个过程可以通过关闭 stdout-chan 来终止。如果你不想关闭stdout-chan，你可以创建一个中间通道，将值放到stdout-chan，直到你通过pipe关闭它。
非常感谢！我刚刚愉快地从我对line-seq 的调用中删除了(try … catch)es，并将它们替换为onto-chan。为了避免alts!! 的复杂性，我将每个(loop [] …) 替换为(while (not @terminated?) …)。每个线程现在只有 3-4 行日志。