Clojure/Java：在对 Amazon S3 数据流执行复杂操作时最小化带宽消耗的最有效方法

Question

我正在使用 BufferedReader 执行对象的流读取。

我需要对这个对象做两件事：

1. 将其传递给 SuperCSV csv 阅读器
2. 获取原始行并将它们保留在（Clojure）延迟序列中

目前，我必须使用两个不同的 BufferedReader：一个作为 SuperCSV CSV 的参数reader 类和一个用于初始化原始行的惰性序列的类。我实际上下载了 S3 对象两次，这是昂贵的 ($) 并且速度慢。

我的一位同事指出，我正在寻找类似于 Unix“tee”命令的东西。可以以某种方式“拆分”、下载一大块数据并将副本传递给惰性序列和 csv 读取器功能的 BufferedReader 将很有用。

我目前还在研究是否可以将惰性序列包装在 BufferedReader 中并将该传递给超级 csv。在将非常大的惰性序列传递给多个使用者时，我遇到了一些 Java 堆空间问题，因此我对采用此解决方案感到有点担心。

另一个解决方案是在本地下载文件，然后在该文件上打开两个流。这消除了流式传输背后的原始动机：允许在数据开始到达时立即开始处理文件。

最后的解决方案是实现我自己的 CSV 阅读器，它返回已解析的 CSV 和原始未解析的行，也是只有在其他方法都不起作用的情况下我才会考虑的解决方案。如果您使用过非常可靠的 CSV 阅读器，它可以返回已解析的 CSV 数据的 Java 哈希值和原始未解析的行，请告诉我！

谢谢！

Answer 1

我倾向于从网络创建一系列行，然后将其交给需要处理该序列的许多进程；持久数据结构在这方面很酷。在需要将字符串序列转换为 Reader 并可以将其交给 SuperCSV api 的情况下，这似乎可行：

(import '[java.io Reader StringReader])

(defn concat-reader
  "Returns a Reader that reads from a sequence of strings."
  [lines]
  (let [srs (atom (map #(StringReader. %) lines))]
    (proxy [Reader] []
      (read 
        ([] 
          (let [c (.read (first @srs))]
            (if (and (neg? c) (swap! srs next))
              (.read this)
              c)))
        ([cbuf] 
          (.read this cbuf 0 (count cbuf)))
        ([cbuf off len]
          (let [actual (.read (first @srs) cbuf off len)]
            (if (and (neg? actual) (swap! srs next))
              (.read this cbuf off len)
              actual))))
      (close [] ))))

例如

user=> (def r (concat-reader ["foo" "bar"]))
#'user/r
user=> (def cbuf (char-array 2))
#'user/cbuf
user=> (.read r cbuf)
2
user=> (seq cbuf)
(\f \o)
user=> (char (.read r))
\o
user=> (char (.read r))
\b

Answer 2

解决方案是使用单个 BufferedReader 进行所有访问，然后在每次将其传递到需要从头开始读取的功能时重置它。