果心异步循环被阻止,等待从通道读取

假设我有一个频道 out (chan) . 我需要获取放入通道的值并将其添加。值的数量尚未确定(因此无法使用带有 (<! out) )来自外部IO。我用的是固定的 timeout 具有 alts! 但这似乎不是解决问题的最佳方式。到目前为止,我有以下内容(我从 https://gist.github.com/schaueho/5726a96641693dce3e47 )

(go-loop
      [[v ch] (alts! [out (timeout 1000)])
       acc 0]
      (if-not v
        (do (close! out)
            (deliver p acc))
        (do
          (>! task-ch (as/progress-tick))
          (recur (alts! [out (timeout 1000)]) (+ acc v)))))

我遇到的问题是,超时1000有时是不够的,会导致go循环过早退出(因为IO操作可能需要超过1000毫秒才能完成并将val放入 out 通道)。我认为增加超时值不是一个好主意,因为它可能会导致我等待的时间过长。

确保从输出通道读取所有数据并从循环中正确退出的最佳方法是什么?

更新:

为什么要使用超时? 因为放入通道中的值的数量不是固定的;也就是说,我无法创建退出案例。在没有出口的情况下,go循环将停在等待中( (<!out) )用于输入通道输出的值。如果你有一个没有超时的解决方案,那就太棒了。

如何知道已读取最后一个值? 我没有。这就是问题所在。这就是为什么我要使用超时和alts!!退出go循环。

考虑到结果,你想做什么? 现在只需简单的添加。然而,这并不重要。

更新最终版本:

我想出了一种方法来获得我要处理的值的数量。所以我修改了我的逻辑来利用它。我仍然要使用超时和alts!防止任何锁定。

(go-loop
     [[v _] (alts! [out (timeout 1000)])
      i 0
      acc 0]
      (if (and v (not= n i))
        (do
          (>! task-ch (as/progress-tick))
          (recur (alts! [out (timeout 1000)]) (inc i) (+ acc v)))
        (do (close! out)
            (deliver p* (if (= n i) acc nil)))))