克洛朱尔.Http 流文件

Question

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

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

任务是"流式传输"带有块的文件，以避免内存不足错误。

Answer 1

在get-file函数中，您根据byte-seq的结果调用byte-array。byte-array将意识到返回的LazySeqbyte-seq这意味着所有这些都将在内存中。

据我所知(至少对于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符合戒指规格，: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))