确保所有请求完成后关闭 clj-http 连接管理器的正确方法

Question

我的代码是clj-http、core.async 设施和atom 的组合。它创建了一些线程来获取和解析一堆页面：

(defn fetch-page
  ([url] (fetch-page url nil))
  ([url conn-manager]
    (-> (http.client/get url {:connection-manager conn-manager})
        :body hickory/parse hickory/as-hickory)))

(defn- create-worker
  [url-chan result conn-manager]
  (async/thread
    (loop [url (async/<!! url-chan)]
      (when url
        (swap! result assoc url (fetch-page url conn-manager))
        (recur (async/<!! url-chan))))))

(defn fetch-pages
  [urls]
  (let [url-chan (async/to-chan urls)
        pages (atom (reduce (fn [m u] (assoc m u nil)) {} urls))
        conn-manager (http.conn-mgr/make-reusable-conn-manager {})
        workers (mapv (fn [_] (create-worker url-chan pages conn-manager))
                      (range n-cpus))]
    ; wait for workers to finish and shut conn-manager down
    (dotimes [_ n-cpus] (async/alts!! workers))
    (http.conn-mgr/shutdown-manager conn-manager)

    (mapv #(get @pages %) urls)))

这个想法是使用多个线程来减少获取和解析页面的时间，但我想不使服务器超载，一次发送大量请求 - 这就是为什么使用了连接管理器。我不知道我的方法是否正确，欢迎提出建议。目前的问题是最后一个请求失败，因为连接管理器在它们终止之前被关闭：Exception in thread "async-thread-macro-15" java.lang.IllegalStateException: Connection pool shut down。

主要问题：如何在正确的时刻关闭连接管理器（以及为什么我当前的代码无法关闭）？支线任务：我的方法对吗？如果没有，我可以做些什么来一次获取和解析多个页面，同时又不会使服务器过载？

谢谢！

Answer 1

问题是async/alts!! 返回第一个结果（并且会继续这样做，因为workers 永远不会改变）。我认为使用async/merge 建立一个频道，然后反复读取它应该可以工作。

(defn fetch-pages
  [urls]
  (let [url-chan (async/to-chan urls)
        pages (atom (reduce (fn [m u] (assoc m u nil)) {} urls))
        conn-manager (http.conn-mgr/make-reusable-conn-manager {})
        workers (mapv (fn [_] (create-worker url-chan pages conn-manager))
                      (range n-cpus))
        all-workers (async/merge workers)]
    ; wait for workers to finish and shut conn-manager down
    (dotimes [_ n-cpus] (async/<!! all-workers))
    (http.conn-mgr/shutdown-manager conn-manager)

    (mapv #(get @pages %) urls)))

或者，您可以重复并继续缩小workers，以便您只等待以前未完成的工人。

(defn fetch-pages
  [urls]
  (let [url-chan (async/to-chan urls)
        pages (atom (reduce (fn [m u] (assoc m u nil)) {} urls))
        conn-manager (http.conn-mgr/make-reusable-conn-manager {})
        workers (mapv (fn [_] (create-worker url-chan pages conn-manager))
                      (range n-cpus))]
    ; wait for workers to finish and shut conn-manager down
    (loop [workers workers]
      (when (seq workers)
        (let [[_ finished-worker] (async/alts!! workers)]
          (recur (filterv #(not= finished-worker %) workers)))))

    (http.conn-mgr/shutdown-manager conn-manager)    
    (mapv #(get @pages %) urls)))

Answer 2

我相信 Alejandro 关于你的错误原因是正确的，这是合乎逻辑的，因为你的错误表明你在所有请求完成之前关闭了连接管理器，所以很可能所有工作人员都没有完成你把它关了。

我将提出的另一个解决方案源于这样一个事实，即您实际上并没有在您的 create-worker 线程中做任何需要它成为通道的事情，该通道是由 async/thread 隐式创建的。因此，您可以将其替换为 future，如下所示：

(defn- create-worker
  [url-chan result conn-manager]
  (future
    (loop [url (a/<!! url-chan)]
      (when url
        (swap! result assoc url (fetch-page url conn-manager))
        (recur (a/<!! url-chan))))))

在您的 fetch-pages 函数中，通过取消引用“加入”：

(doseq [worker workers]
  @worker) ; alternatively, use deref to specify timeout 

这消除了很多 core.async 对非 core.async 问题的干扰。这当然取决于您保持收集数据的方法不变，即在原子上使用swap! 来跟踪页面数据。如果您要将fetch-page 的结果发送到返回通道或类似的东西，那么您需要保留当前的thread 方法。

关于您对服务器超载的担忧——您尚未定义“超载”服务器的含义。这有两个维度：一个是请求的rate（例如每秒的请求数），另一个是并发请求的数量。您当前的应用程序具有n 工作线程，这就是有效的并发性（以及连接管理器中的设置）。但这无助于解决每秒请求的速率。

这比看起来要复杂一些，尽管这是可能的。您必须考虑每单位时间所有线程完成的所有请求的总数，并且管理这不是在这里的一个答案中要解决的问题。我建议您对节流和速率限制进行一些研究，并尝试一下，然后从那里提出问题。