Clojure。 HTTP流文件

Question

我想流式传输大型二进制文件(exe、jpg、...，所有类型的文件)。看来aleph客户端可以做到。 我看了官方示例，了解到如果我们将惰性序列传递给正文，则响应可以传递一个流。

(defn streaming-numbers-handler
  "Returns a streamed HTTP response, consisting of newline-delimited numbers every 100
   milliseconds.  While this would typically be represented by a lazy sequence, instead we use
   a Manifold stream.  Similar to the use of the deferred above, this means we don't need
   to allocate a thread per-request.
   In this handler we're assuming the string value for `count` is a valid number.  If not,
   `Integer.parseInt()` will throw an exception, and we'll return a `500` status response
   with the stack trace.  If we wanted to be more precise in our status, we'd wrap the parsing
   code with a try/catch that returns a `400` status when the `count` is malformed.
   `manifold.stream/periodically` is similar to Clojure's `repeatedly`, except that it emits
   the value returned by the function at a fixed interval."
  [{:keys [params]}]
  (let [cnt (Integer/parseInt (get params "count" "0"))]
    {:status 200
     :headers {"content-type" "text/plain"}
     :body (let [sent (atom 0)]
             (->> (s/periodically 100 #(str (swap! sent inc) "\n"))
               (s/transform (take cnt))))}))

我有以下代码:

(ns core
  (:require [aleph.http :as http]
            [byte-streams :as bs]
            [cheshire.core :refer [parse-string generate-string]]
            [clojure.core.async :as a]
            [clojure.java.io :as io]
            [clojure.string :as str]
            [clojure.tools.logging :as log]
            [clojure.walk :as w]
            [compojure.core :as compojure :refer [ANY GET defroutes]]
            [compojure.route :as route]
            [me.raynes.fs :as fs]
            [ring.util.response :refer [response redirect content-type]]
            [ring.middleware.params :as params]
            [ring.util.codec :as cod])

  (:import java.io.File)
  (:import java.io.FileInputStream)
  (:import java.io.InputStream)
  (:gen-class))


(def share-dir "\\\\my-pc-network-path")

(def content-types
  {".png"  "image/png"
   ".GIF"  "image/gif"
   ".jpeg" "image/jpeg"
   ".svg"  "image/svg+xml"
   ".tiff" "image/tiff"
   ".ico"  "image/vnd.microsoft.icon"
   ".bmp"  "image/vnd.wap.wbmp"
   ".css"  "text/css"
   ".csv"  "text/csv"
   ".html" "text/html"
   ".txt"  "text/plain"
   ".xml"  "text/xml"})

(defn byte-seq [^InputStream is size]
  (let [ib (byte-array size)]
    ((fn step []
       (lazy-seq
         (let [n (.read is ib)]
           (when (not= -1 n)
             (let [cb (chunk-buffer size)]
               (dotimes [i size] (chunk-append cb (aget ib i)))
               (chunk-cons (chunk cb) (step))))))))))

(defn get-file [req]
  (let [uri (:uri req)
        file (str/replace (w/keywordize-keys (cod/form-decode uri)) #"/file/" "")
        full-dir (io/file share-dir file)
        _ (log/debug "FULL DIR: "full-dir)
        filename (.getName (File. (.getParent full-dir)))
        extension (fs/extension filename)
        _ (log/debug "EXTENSION: " extension)
        resp {:status  200
              :headers {"Content-type"        (get content-types (.toLowerCase extension) "application/octet-stream")
                        "Content-Disposition" (str "inline; filename=\"" filename "\"")}
              :body    (with-open [is (FileInputStream. full-dir)]
                         (let [bs (byte-seq is 4096)]
                           (byte-array bs)))}
        ]
    resp))

(def handler
  (params/wrap-params
    (compojure/routes
      (GET "/file/*"         [] get-file)
      (route/not-found "No such file."))))


(defn -main [& args]
  (println "Starting...")
  (http/start-server handler {:host "0.0.0.0" :port 5555}))

我得到 uri 并尝试读取 block 文件。我想这样做是因为文件可能有 3 GB 左右。所以，我期望的是应用程序不会使用超过 block 大小的内存。 但是当我为应用程序设置 1GB(-Xmx 选项)时，它占用了所有内存(1 GB)。 为什么要占用1GB？ JVM 是这样工作的吗？ 当我有 100 个同时连接时(例如，每个文件为 3GB)，我得到 OutOfMemoryError。

任务是“流式传输”带有 block 的文件以避免 OutOfMemoryError。

Answer 1

在 get-file 函数中，您正在对 byte-seq 的结果调用 byte-array。 byte-array 将实现 byte-seq 返回的 LazySeq 意味着所有这些都将在内存中。

据我所知(至少对于 TCP 服务器)，aleph 接受 ByteBuffer 的任何惰性序列作为主体，并会为您优化处理它，因此您可以只返回调用 byte-seq< 的结果

(defn get-file [req]
  (let [uri (:uri req)
        file (str/replace (w/keywordize-keys (cod/form-decode uri)) #"/file/" "")
        full-dir (io/file share-dir file)
        _ (log/debug "FULL DIR: "full-dir)
        filename (.getName (File. (.getParent full-dir)))
        extension (fs/extension filename)
        _ (log/debug "EXTENSION: " extension)
        resp {:status  200
              :headers {"Content-type"        (get content-types (.toLowerCase extension) "application/octet-stream")
                        "Content-Disposition" (str "inline; filename=\"" filename "\"")}
              :body    (with-open [is (FileInputStream. full-dir)]
                         (byte-seq is 4096))}
        ]
    resp))

备选

Aleph 符合 ring specification和 :body 接受 File 或 InputStream。所以不需要自己返回字节。

(defn get-file [req]
  (let [uri (:uri req)
        file (str/replace (w/keywordize-keys (cod/form-decode uri)) #"/file/" "")
        full-dir (io/file share-dir file)
        _ (log/debug "FULL DIR: "full-dir)
        filename (.getName (File. (.getParent full-dir)))
        extension (fs/extension filename)
        _ (log/debug "EXTENSION: " extension)
        resp {:status  200
              :headers {"Content-type"        (get content-types (.toLowerCase extension) "application/octet-stream")
                        "Content-Disposition" (str "inline; filename=\"" filename "\"")}
              :body    full-dir}
        ]
    resp))