基于通道的懒序列

Question

我正在尝试将通道转换为 seq。这是我目前使用的

(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机器上的Linux系统上的OpenJDK）。

希望这有助于您。

Comments

不必是1000，20也行。我已经更新了问题，并添加了示例。

---

哦，如果是那样的话，‘go’就是罪魁祸首。‘go’使用它自己的线程池，且线程数量并不多。但是！因为读取是阻塞的，这里根本不需要‘go’。请参见我链接的实现，它支持大量线程。

---

如果我想在go块中或ClojureScript（ClojureScript没有阻塞操作）中使用这个函数怎么办呢？

---

在这种情况下，可能有点复杂。你可能需要了解有关问题的更多信息（例如，出现在通道中的项目的模式/时间/频率），然后使用这些信息来构建非阻塞的、可能可重新启动的具有超时或类似的代码。在许多情况下，JavaScript仍然是有效的单线程。