如何在clojure中按比例产生一个惰性序列?

问题描述

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

I have a database server and I fetch data from it. Sometimes data have millions rows and more so I use laziness for downloading. I use Server Side Cursors from clojure.jdbc library https://funcool.github.io/clojure.jdbc/latest/#cursor-queries to fetch data lazily.

现在我有问题.我需要从一个惰性序列中产生初始的500个元素，然后该程序必须等待10分钟才能收到一个信号，该信号向程序报告会产生下一个500个元素，依此类推，直到我从服务器接收到所有数据.但是，如果10分钟后仍未收到报告，则程序必须关闭连接.

Now I have a problem. I need produce initial 500 elements from a lazy-sequence, then the program must wait for 10 minutes to get a signal which report to the program produce next 500 elements and so on until i receive all data from server. But if a report didn`t arrive for 10 minutes, the program must close a connection.

我写了样本:

(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)))

您能帮我使用哪些Clojure工具来解决我的问题吗?我假设我应该使用core.async为循环创建通道.我对吗?以及我该如何创建一个函数，使其在接收到适当的信号后停止执行循环10分钟或继续执行?

Can you help what clojure tools I should use to solve my problems? I assume I should use core.async to create channels for loops. Am I right? And how should I create function which stop executing loop for 10 minutes or continue executing if I receive appropriate signal?

推荐答案

实际上，您应该为此使用core.async，它们具有超时通道用于超时目的，并且具有替代！，您可以等待"一个值，以从任意数量的渠道中弹出.完整的代码虽然有点复杂.输出行将推送"到输出通道.

Indeed you should use core.async for this, they have timeout channel for the timeout purpose and with alt! you can "wait" for a value to pop out from any number of channels. The complete code will be a bit complex though. Output rows would be "pushed" to an output channel.

据我所知，with-open和惰性序列的组合不好，因为游标会过早关闭.但是我对clojure.jdbc库不熟悉.

To my knowledge with-open and lazy sequences don't compose nicely, as the cursor would be closed prematurely. But I am not familiar with clojure.jdbc library.

这篇关于如何在clojure中按比例产生一个惰性序列?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！