当一个线程不断从通道中取出时，通道什么时候会被丢弃？

Question

考虑以下取自example walkthrough of core.async 的代码：

(let [c1 (chan)
      c2 (chan)]
   (thread 
      (while true
         (let [[v ch] (alts!! [c1 c2])]
              (println "Read" v "from" ch))))
   (>!! c1 "hi")
   (>!! c2 "there"))

我的假设是线程引用了c1 和c2 两个通道，并且基本上会永远运行，试图从其中一个永远不会出现的值中获取值。因此，通道既不会被垃圾收集，线程也不会终止。即使我们明确地close! 通道，线程仍然会继续。我的结论是正确的还是我遗漏了什么？

我之所以这么问，是因为我正试图找到一种方法，让我能够成功地使用这样一个无休止运行的消费者来测试这样的 core.async 代码。我当前的尝试如下所示：

(let [c1 (chan)
      c2 (chan)]
    (go 
        (>!! c1 "hi")
        (>!! c2 "there"))
    (async/thread
      (loop [[v ch] (alts!! [c1 c2])]
        (println "Read" v "from" ch)
        (when-let [[nv nch] (alts!! [c1 c2])]
          (if nv
             (recur [nv nch])
             :done)))))

这会返回一个结果通道（来自thread），我想阻止它获取:done 值，但我需要一种关闭（至少一个）通道的方法。我可以返回两个频道的列表c1, c2 和thread 返回的结果频道，然后close! 例如c1 之后查看结果通道，但这非常难看：

(let [c1 (chan)
      c2 (chan)]
    (go 
        (>!! c1 "hi")
        (>!! c2 "there"))
    [c1 c2 (async/thread
      (loop [[v ch] (alts!! [c1 c2])]
        (println "Read" v "from" ch)
        (when-let [[nv nch] (alts!! [c1 c2])]
          (if nv
              (recur [nv nch])
              :done))))])
=> [#<ManyToManyChannel clojure.core.async.impl.channels.ManyToManyChannel@60eb5def> #<ManyToManyChannel clojure.core.async.impl.channels.ManyToManyChannel@7c64279e> #<ManyToManyChannel clojure.core.async.impl.channels.ManyToManyChannel@136535df>]
   Read hi from #<ManyToManyChannel clojure.core.async.impl.channels.ManyToManyChannel@60eb5def>
   Read there from #<ManyToManyChannel clojure.core.async.impl.channels.ManyToManyChannel@7c64279e>

(let [[c1 c2 resultchan] *1]
  (close! c1)
  (<!! resultchan))
=>:done

或者，我可能会发送一个特殊的“通信结束”值，然后我可以在接收端进行检查。

这方面的最佳做法是什么样的？

Answer 1

我不知道这种特殊情况是否有最佳实践之类的东西。这是我对这个问题的解决方案。我认为这很简单。

(defn alts-while-open [f & chans]
   (let [a-chans (atom (set chans))]
     (go (while (< 0 (count @a-chans))
           (println "iteration started : " (vec @a-chans))
           (let [[v ch] (alts! (vec @a-chans))]
             (if v
               (f v ch)
               (swap! a-chans #(disj % ch))))))))

函数 f 在 alts 的结果上执行！在通道打开的情况下。我在这里保留了一个带有一组开放通道的原子。一旦我找到一个关闭的频道，我就把它从这个集合中删除。如果没有更多打开的通道，while 循环将停止。你可以运行它：

(def c1 (chan))
(def c2 (chan))
(def c3 (chan))
(alts-while-open (fn [v ch] (println v)) c1 c2 c3)

现在，当将某些内容写入这些通道中的任何一个时，它就会被打印出来。您可以看到之后开始新的迭代。关闭通道后，您可以看到迭代开始但通道向量减少了。一旦所有通道都关闭，while 循环就会停止。

很难回答是否使用关闭的问题！函数或其他一些通知机制来停止循环。我认为这取决于情况。如果对“通信结束”没有任何复杂的处理，我会关闭！这个频道。如果有更复杂的逻辑 - 例如有成功的“通信结束”和失败的“通信结束”选项，我想以不同的方式处理它们，那么我宁愿发送一条特殊消息。像

[:end-of-communication :success]

Answer 2

当您不发送简单的值时，发送特殊的end-of-communication 的想法不起作用，因为您不能保证值会按照您想要的顺序放置和获取。

下一个想法是发送者和接收者都事先知道要处理的值的数量，例如像这样：

user> (<!!
        (let [c1 (chan)
              values ["hi" "there"]
              vcount (count values)]
           (doseq [value values]
             (thread
                  (>!! c1 value)))
           (thread
               (loop [recvalue (<!! c1)
                      reccount 1]
                  (println "Read" recvalue)
                  (if (= reccount vcount)
                      (do (close! c1)
                          :done)
                      (recur (<!! c1) (inc reccount)))))))
Read hi
Read there
:done

这原则上可行，但有一个明显的缺点，即您必须在设置发送和接收流程之前就金额达成一致，而不太明显的缺点是如果发送方出现问题，您将结束再次在接收端无休止地等待（假设发送端不仅仅将值放入通道）。

我得出的结论是，使这个可靠的唯一方法是使用timeout 频道，像这样：

(<!! (let [c1 (chan)
           tchan (timeout 1000) 
           values ["hi" "there"]]
       (doseq [value values]
         (thread
           (>!! c1 value)))
       (thread
          (loop [[recvalue rchan] (alts!! [c1 tchan])
                 timeoutchan tchan]
            (if (= rchan timeoutchan)
                (do (close! c1)
                    :done)
                (do (println "Read" recvalue)
                    (let [newtimeout (timeout 1000)]
                        (recur (alts!! [c1 newtimeout])
                               newtimeout))))))

Read hi
Read there
:done