jdbc – 如何在clojure中按部分生成一个惰性序列？

2019年7月28日 208次阅读

我有一个数据库服务器,我从中获取数据.有时数据有数百万行甚至更多,所以我使用懒惰进行下载.我使用clojure.jdbc库
https://funcool.github.io/clojure.jdbc/latest/#cursor-queries中的服务器端游标来懒惰地获取数据.

现在我有一个问题.我需要从延迟序列生成初始500个元素,然后程序必须等待10分钟才能获得一个信号,该信号向程序生成下一个500个元素,依此类推,直到我从服务器接收所有数据.但如果报告没有到达10分钟,程序必须关闭连接.

我写了样本：

(def lazyseq_maps (atom {:seq_1 {:next_500 false :data nil} :seq_2 {:next_500 false :data nil}})) ; here is a collection of all unfinished lazy sequences waiting for signal to continue produce elements

(jdbc/atomic conn
 (with-open [cursor (jdbc/fetch-lazy conn sql]
   (let [lazyseq (jdbc/cursor->lazyseq cursor)]
     (swap! lazyseq_maps assoc seq_id {:next_500 true :data nil})
     (loop [lazyseq_rest lazyseq
            count 1]
            (if (:next_500 (seq_id @lazyseq_maps))
              (do
                (swap! lazyseq_maps update-in [seq_id :data] conj (first lazyseq_rest))
                (when (= 0 (mod count 500))
                  (swap! lazyseq_maps assoc-in [seq_id :next_500] false))
                (recur (rest lazyseq) (inc count)))
              ;
              (func-for-waiting-signal)))) ; here I don`t know how to create function waiting signal to continue fetching data
   (seq_id @lazyseq_maps)))

你能帮我解决我的问题吗？我假设我应该使用core.async为循环创建通道.我对吗？如果我收到适当的信号,我应该如何创建停止执行循环10分钟或继续执行的功能？

最佳答案 实际上你应该使用core.async,它们有
timeout通道用于超时目的,而
alt!你可以“等待”从任意数量的通道中弹出一个值.完整的代码虽然有点复杂.输出行将“推送”到输出通道.

据我所知,开放和懒惰的序列不能很好地组合,因为光标会过早关闭.但我不熟悉clojure.jdbc库.

点赞