Clojurescript：使用核心/异步通道分块处理请求

Posted 2023-03-14

【中文标题】Clojurescript：使用核心/异步通道分块处理请求

【英文标题】：Clojurescript: process requests in chunks with core/async channels

【发布时间】：2018-10-26 21:24:06

【问题描述】：

我有以下场景：

我使用一些服务来检索一些数据，并将其传递给我的输入。

有一些输入参数，我需要对上述服务执行 N 个请求，收集输出并为每个输出执行一些 CPU 繁重的任务。

我正在尝试使用核心/异步通道来实现这一点。

这是我的尝试（示意性地），它有点工作，但它的行为不是我想要的。 将不胜感激任何关于如何改进它的提示。

(defn produce-inputs
  [in-chan inputs]
  (let input-names-seq (map #(:name %) inputs)]
    (doseq [input-name input-names-seq]
      (async/go
        (async/>! in-chan input-name)))))

(defn consume
  [inputs]
  (let [in-chan (async/chan 1)
        out-chan (async/chan 1)]
        (do
          (produce-inputs in-chan inputs)
          (async/go-loop []
                   (let [input-name (async/<! in-chan)]
                     (do
                         (retrieve-resource-from-service input-name 
                                                         ; response handler
                                                         (fn [resp]
                                                           (async/go
                                                             (let [result (:result resp)]
                                                               (async/>! out-chan result)))))
                         (when input-name
                           (recur)))))

     ; read from out-chan and do some heavy work for each entry
     (async/go-loop []
                   (let [result (async/<! out-chan)]
                         (do-some-cpu-heavy-work result))))))

; entry point
(defn run
  [inputs]
  (consume inputs))

有没有什么方法可以更新它，以便每时每刻都有不超过五个服务请求 (retrieve-resource-from-service) 处于活动状态？

如果我的解释不清楚，请提出问题，我会更新它。

【参考方案1】：

您可以创建另一个通道作为令牌桶来限制您的请求速率。

See this link 使用令牌桶进行每秒速率限制的示例。

要限制同时请求的数量，您可以执行以下操作：

(defn consume [inputs]
  (let [in-chan (async/chan 1)
        out-chan (async/chan 1)
        bucket (async/chan 5)]
    ;; ...
    (dotimes [_ 5] (async/put! bucket :token))
    (async/go-loop []
      (let [input-name (async/<! in-chan)
            token (async/<! bucket)]
        (retrieve-resource-from-service
          input-name 
          ; response handler
          (fn [resp]
            (async/go
              (let [result (:result resp)]
                (async/>! out-chan result)
                (async/>! bucket token)))))
        (when input-name
          (recur))))
    ;; ...
    ))

创建了一个新频道bucket，并将五个项目放入其中。在触发请求之前，我们从存储桶中取出一个令牌，并在请求完成后将其放回原处。如果bucket 频道中没有令牌，我们必须等到其中一个请求完成。

注意：这只是代码的草图，您可能需要更正它。特别是，如果您的 retrieve-resource-from-service 函数中有任何错误处理程序，您也应该将令牌放回原处以防出现错误以避免最终死锁。

【讨论】：

谢谢。我喜欢这种方法，会试一试并回信。