4

我有一个代码，它是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.

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

谢谢！

4

2 回答 2

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)))
于 2017-03-08T19:07:15.303 回答
1

我相信 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方法。

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

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

于 2017-03-08T20:05:27.720 回答