web-services - Clojure 中 Web 服务的异步作业队列

Question

目前，我正在尝试使用 RESTful API 构建一个 Web 服务，该 API 可以处理一些长时间运行的任务(作业)。

这个想法是用户通过执行 POST 提交作业，该 POST 返回一些用于检查作业状态的 URL，其中还包含结果的 URL。一旦作业完成(即某些值已写入数据库)，结果 URL 将返回适当的信息(而不是没有结果)，作业 URL 将指示已完成状态。

不幸的是，计算非常密集，因此一次只能运行一个，因此需要对作业进行排队。

在伪中需要这样的东西

(def job-queue (atom queue)) ;; some queue 
(def jobs (atom {}))

(defn schedule-job [params] 
  ;; schedules the job into the queue and 
  ;; adds the job to a jobs map for checking status via GET
  ;; note that the job should not  be evaluated until popped from the queue
)

(POST "/analyze" [{params :params}] 
 (schedulde-job params))

(GET "job/:id" [:d] 
 (get @jobs id))

;; Some function that pops the next item from the queue 
;; and evaluates it when the previous item is complete
;; Note: should not terminate when queue is empty! 

我查看了 Lamina它允许异步处理，但它似乎不适合我的需要。

我的问题是如何在前一个完成后使作业队列出队并执行其任务，而不会在队列为空时终止，即永久处理传入的作业。

Answer 1

java.util.concurrent.ExecutorService 可能就是您想要的。这允许您提交作业以供以后执行，并返回一个 Future，您可以查询它以发现它是否已完成。

(import '[java.util.concurrent Callable Executors])

(def job-executor
  (Executors/newSingleThreadExecutor))

(def jobs (atom {}))

(defn submit-job [func]
  (let [job-id   (str (java.util.UUID/randomUUID))
        callable (reify Callable (call [_] (func))]
    (swap! jobs assoc job-id (.submit job-executor callable))
    job-id))

(use 'compojure.core)

(defroutes app
  (POST "/jobs" [& params]
    (let [id (submit-job #(analyze params))]
      {:status 201 :headers {"Location" (str "/jobs/" id)}}))
  (GET "/jobs/:id" [id]
    (let [job-future (@jobs id)]
      (if (.isDone job-future)
        (.get job-future)
        {:status 404}))))