10

我正在使用 Clojure 应用程序从 Web API 访问数据。我将发出大量请求，其中许多请求将导致发出更多请求，因此我想将请求 URL 保留在队列中，以便在后续下载之间留出 60 秒。

在这篇博文之后，我把它放在一起：

(def queue-delay (* 1000 60)) ; one minute

(defn offer!
  [q x]
  (.offerLast q x)
  q)

(defn take!
  [q]
  (.takeFirst q))

(def my-queue (java.util.concurrent.LinkedBlockingDeque.))

(defn- process-queue-item
  [item]
  (println ">> " item)   ; this would be replaced by downloading `item`
  (Thread/sleep queue-delay))

如果我(future (process-queue-item (take! my-queue)))在代码中的某处包含 a ，那么我可以在 REPL 中(offer! my-queue "something")看到“>> something”立即打印出来。到现在为止还挺好！但是我需要队列在我的程序处于活动状态的整个过程中持续存在。我刚刚提到的(future ...)调用可以将一个项目从队列中拉出，一旦它可用，但我想要一些能够持续观察队列并process-queue-item在有可用时调用的东西。

此外，与 Clojure 通常对并发的热爱相反，我想确保一次只发出一个请求，并且我的程序等待 60 秒以发出每个后续请求。

我认为这个 Stack Overflow 问题是相关的，但我不确定如何调整它来做我想做的事情。如何连续轮询我的队列并确保一次只运行一个请求？

4

3 回答 3

3

这是我为乐趣而做的一个项目的代码片段。它并不完美，但可以让您了解我是如何解决“等待 55 秒等待第一项”问题的。它基本上循环通过承诺，使用期货立即处理事情或直到承诺“变得”可用。

(defn ^:private process
  [queues]
  (loop [[q & qs :as q+qs] queues p (atom true)]
    (when-not (Thread/interrupted)
      (if (or
            (< (count (:promises @work-manager)) (:max-workers @work-manager))
            @p) ; blocks until a worker is available
        (if-let [job (dequeue q)]
          (let [f (future-call #(process-job job))]
            (recur queues (request-promise-from-work-manager)))
          (do
            (Thread/sleep 5000)
            (recur (if (nil? qs) queues qs) p)))
        (recur q+qs (request-promise-from-work-manager))))))

也许你可以做类似的事情？代码不是很好，可能需要重新编写才能使用lazy-seq，但这只是我还没有做的练习！

于 2012-08-29T16:20:23.970 回答
1

这很可能很疯狂，但你总是可以使用这样的函数来创建一个慢下来的惰性序列：

(defn slow-seq [delay-ms coll]
  "Creates a lazy sequence with delays between each element"
  (lazy-seq 
    (if-let [s (seq coll)]
        (do 
          (Thread/sleep delay-ms)
          (cons (first s)
                (slow-seq delay-ms (rest s)))))))

这将基本上确保每个函数调用之间的延迟。

您可以将其与以下内容一起使用，以毫秒为单位提供延迟：

(doseq [i (slow-seq 500 (range 10))]
  (println (rand-int 10))

或者，您可以将函数调用放在序列中，例如：

(take 10 (slow-seq 500 (repeatedly #(rand-int 10))))

显然，在上述两种情况下，您都可以替换(rand-int 10)为用于执行/触发下载的任何代码。

于 2012-09-04T07:52:03.330 回答
1

我最终推出了自己的小型库，我称之为simple-queue。你可以在 GitHub 上阅读完整的文档，但这里是完整的源代码。我不会更新这个答案，所以如果你想使用这个库，请从 GitHub 获取源代码。

(ns com.github.bdesham.simple-queue)

(defn new-queue
  "Creates a new queue. Each trigger from the timer will cause the function f
  to be invoked with the next item from the queue. The queue begins processing
  immediately, which in practice means that the first item to be added to the
  queue is processed immediately."
  [f & opts]
  (let [options (into {:delaytime 1}
                      (select-keys (apply hash-map opts) [:delaytime])),
        delaytime (:delaytime options),
        queue {:queue (java.util.concurrent.LinkedBlockingDeque.)},
        task (proxy [java.util.TimerTask] []
               (run []
                 (let [item (.takeFirst (:queue queue)),
                       value (:value item),
                       prom (:promise item)]
                   (if prom
                     (deliver prom (f value))
                     (f value))))),
        timer (java.util.Timer.)]
    (.schedule timer task 0 (int (* 1000 delaytime)))
    (assoc queue :timer timer)))

(defn cancel
  "Permanently stops execution of the queue. If a task is already executing
  then it proceeds unharmed."
  [queue]
  (.cancel (:timer queue)))

(defn process
  "Adds an item to the queue, blocking until it has been processed. Returns
  (f item)."
  [queue item]
  (let [prom (promise)]
    (.offerLast (:queue queue)
                {:value item,
                 :promise prom})
    @prom))

(defn add
  "Adds an item to the queue and returns immediately. The value of (f item) is
  discarded, so presumably f has side effects if you're using this."
  [queue item]
  (.offerLast (:queue queue)
              {:value item,
               :promise nil}))

使用此队列返回值的示例：

(def url-queue (q/new-queue slurp :delaytime 30))
(def github (q/process url-queue "https://github.com"))
(def google (q/process url-queue "http://www.google.com"))

调用将阻塞，因此两个语句q/process之间会有 30 秒的延迟。def

使用此队列纯粹用于副作用的示例：

(defn cache-url
  [{url :url, filename :filename}]
  (spit (java.io.File. filename)
        (slurp url)))

(def url-queue (q/new-queue cache-url :delaytime 30))
(q/add url-queue {:url "https://github.com",
                  :filename "github.html"})    ; returns immediately
(q/add url-queue {:url "https://google.com",
                  :filename "google.html"})    ; returns immediately

现在调用q/add立即返回。

于 2012-09-11T17:02:22.677 回答