core.async中的重叠分区

Question

在Clojure中，我可以通过调优step参数到partition来获得集合的重叠分区

(partition 3 1 (range 20))

;; ((0 1 2) (1 2 3) (2 3 4) (3 4 5) ...)

core.async确实有一个分区函数，但是由于它不接受step参数，所以我无法获得重叠分区：

(let [c (chan)]
  (go (doseq [n (range 20)]
        (>! c n)))

  (go-loop [p (async/partition 3 c)]
    (when-let [v (<! p)]
      (prn v)
      (recur p))))

;;[0 1 2]
;;[3 4 5]
;;[6 7 8]

我意识到，拥有这种功能可能意味着能够从一个通道中多次读取相同的值。我还知道，我可以创建自己的函数，从一个通道读取任意数量的值，并构建自己的分区。

但是，我想知道是否有任何方法可以通过core.async提供的核心API来实现这一点。

PS。sliding-buffer没有做到这一点，因为我不能同时查看整个缓冲区。

Answer 1

这样做的一种方法是创建一个函数，从通道读取、缓冲值并放入一个新的通道。但我不知道这有多习以为常。

例如，只要从输入通道读取了所需的put!项，下面的函数就会将一个向量放到输出通道中，在每个输出之后跳过step项。

(defn stepped-partition [in n step]
  (let [out (chan)]
    (go-loop [buffer []]
      (when-let [v (<! in)]
        (let [new-buffer (conj buffer v)]
          (if (= (count new-buffer) n)
            (do 
              (put! out new-buffer)
              (recur (subvec new-buffer step)))
            (recur new-buffer)))))
   out))

(def original (chan))
(def partitioned (stepped-partition a 3 2))

(go-loop []
           (when-let [v (<! partitioned)]
             (println v)
             (recur)))

(async/onto-chan original [1 2 3 4 5 6 7 8 9])

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

Answer 2

IMHO，我认为这个愿望

“能够多次从通道读取相同的值”

违反了core.async的原则。

每次从信道读取值时，都要从通道中取出此值。

因此，通道行为的好处是，它保证每个值只读取一次，如果没有值则保证不读取(阻塞/停放线程)，如果通道关闭，则保证为零。

然后，开始解决您的问题的下一个问题应该是:为什么(在core.async上)至少有3个不同的函数在一个通道上/取值。因此，将通信通道视为一个集合，在读者和作者可用之前，有(core.async) 3种不同的应用程序/线程行为：

• 阻塞线程>!! <!!正在运行的线程将被阻塞，直到阅读器和写入器都可用为止。
• 停车线程(使用go宏块) >! <! go块将创建一个伪线程，这个伪线程将被停放，直到阅读器和写入器都可用。此行为不会阻止您的运行线程。
• 异步行为take! put!您只能按写和读的顺序得到保证