基于通道的惰性序列

Question

我正在尝试将一个通道转换成一个序列。这是当前我所使用的

(defn chan->seq [ch]
  (when [v (<!! ch)]
    (lazy-seq (cons v (chan->seq ch)))))

这可以工作，但它使用了一个阻塞式读取，因此当前线程将会被阻塞，直到通道关闭。同时无法有太多的函数调用活跃。有无非阻塞的函数版本？

这个例子演示了这个问题

(doseq [_ (range 20)]
  (go (chan->seq (chan 10))))

(go (println :OK))

最后一个语句将不会打印任何内容，因为 chan→seq 调用堵塞了线程池中的所有线程。如果将 20 更改为 5（在一个新的 repl 会话中），则可正常工作。

Answer 1

你的实现是好的。

我们编写并使用了类似的内容

https://github.com/techascent/tech.parallel/blob/cc1b9b0ef8a0893fe1278a700a4dbb0271f92156/src/tech/parallel.clj#L38-L53

同时无法有 1000 次这个函数的激活调用

对此我并不十分确定。你尝试过了吗？

在资源约束的某些条件下，已停用的线程的堆栈使用可能会引起内存压力，但一般来说，当代调度程序可以很好地处理大量已停用的线程（如果恢复线程操作不是O(1)，我们将会感到惊讶——至少在我们的系统中是这样——我们在AWS的各种EC2机器上使用的OpenJDK Linux系统）。

希望这能有所帮助。

Comments

不一定是1000，20个也可以。我已经在问题中添加了示例。

---

哦，如果是那样的话，`go` 才是罪魁祸首。`go` 使用自己的线程池，其中线程数量不多。但是！因为读取是阻塞的，所以这里根本不需要 `go`。请参阅我提供的实现，它可以与许多线程一起工作。

---

如果我想在go块中使用这个函数怎么办？或者在 clojurescript 中，其中没有阻塞操作？

---

在这种情况下，可能会稍微复杂一点。你可能需要了解问题的其他一些信息（比如项目在通道上出现的模式/时间/频率），然后利用这些信息来构建非阻塞的、可能可重启且带超时的东西，或者类似的。在许多情况下，JavaScript 仍然是有效单线程的。