core.async 中的有状态转换器答案

【问题标题】：Stateful transducers in core.asynccore.async 中的有状态转换器
【发布时间】：2016-01-21 19:33:01
【问题描述】：

我试图了解如何在 core.async 中制作有状态转换器。例如，我将如何制作一个传感器来计算通过通道的元素数量？例如，我想将输入转换为取决于之前出现的对象数量的计数。

从我读到的方法是使用volatile! 来保持传感器内部的状态，但我仍然不确定如何将所有东西放在一起。

【问题讨论】：

计算元素数量到底是什么意思？您的意思是您将一些对象放入通道中，然后呢？就像当您从该通道获取一些 val 时，它应该返回写入它的元素数量？或者像keep-indexed 转换器这样的东西？

【解决方案1】：

您需要一个有状态的转换器返回一个通过 volatile! 跟踪计数而关闭的归约函数。

(defn count-xf [rf]
  (let [ctr (volatile! 0)]
    (fn
      ([] (rf))
      ([result] (rf result))
      ([result _]                         ; we ignore the input as
       (rf result (vswap! ctr inc))))))   ; we just pass on the count

这可以使用核心函数completing来简化

(defn count-xf [rf]
  (let [ctr (volatile! 0)]
    (completing
     (fn [result _]
       (rf result (vswap! ctr inc))))))

E. G。这么用吧

(let [ch (chan 1 count-xf)]
  (onto-chan ch (repeat 10 true))
  (<!! (clojure.core.async/into [] ch)))

;-> [1 2 3 4 5 6 7 8 9 10]

或者，您可以只使用map-indexed 转换器，但这可能会帮助您较少地了解转换器的工作原理。对于这个特定的用例，它还需要一些额外的每一步开销。

(def count-xf (map-indexed (fn [i _] (inc i))))

请注意，它的 implementation 与上面的实现差别不大。

进一步参考：http://clojure.org/reference/transducers

【讨论】：