代码之家 › 专栏 › 技术社区 › Ashwin

在JVM中将数据流式传输给调用者

core.async jvm stream clojure

0

Ashwin · 技术社区 · 6 年前

我有一个函数,它定期获取数据,然后停止获取数据。此函数必须将其定期获取的数据返回给函数的调用者

当它得到
一枪

第二个是一个简单的实现,即你阻止调用者,获取所有数据,然后一次性发送。

但我想实现第一个(我想避免回调)。溪流是这里要用的东西吗?如果是这样,怎么办?如果没有,我怎么回来 something 呼叫者可以查询数据,并在返回没有更多数据的信号时停止?

注意:我在JVM生态系统中,具体来说是clojure。我看了一下clojure库 core.async 哪种方式解决了使用通道的这种问题。但我在想,是否有其他方式可能看起来像这样(假设流是可以使用的)。
Java代码片段

//Function which will periodically fetch MyData until there is no data
public Stream<MyData> myFunction() {
...
}

myFunction().filter(myData -> myData.text.equals("foo"))

0 回复 | 直到 6 年前

1

rmcv 6 年前

也许你可以用 seq -默认情况下,它是懒惰的(如Stream),因此调用者可以决定何时提取数据。以及何时没有更多数据 myFunction 可以简单地结束序列。在执行此操作时,您还可以在其中封装一些优化 myFunction -例如,批量获取数据以尽量减少往返。或者根据您的原始要求定期获取数据。

以下是一个简单的实现:

(defn my-function []
  (let [batch 100]
    (->> (range)
         (map #(let [from (* batch %)
                     to   (+ from batch)]
                  (db-get from to)))
         ;; take while we have data from db-get
         (take-while identity)
         ;; returns as one single seq/Stream
         (apply concat))))

;; use it as a normal seq/Stream
(->> (my-function)
     (filter odd?))

哪里 db-get 大致如下:

(defn db-get [from to]
  ;; return first 1000 records only, i.e. returns nil to signal completion
  (when (< from 1000)
    ;; returns a range of records
    (range from to)))

2

1

Denis Fuenzalida 6 年前

你可能想查一下 https://github.com/ReactiveX/RxJava 和 https://github.com/ReactiveX/RxClojure (似乎不再维持?)