Как надежно выполнять большое количество одновременных HTTPS-запросов в Clojure (/Java)

1. Выполнение 50 HTTP-запросов в секунду, каждый из которых может занимать 200 мс, а это означает, что в любой момент времени может выполняться 100 запросов, звучит как чрезмерное бремя?

Это определенно не избыточно для современного оборудования.

2. Как мне сделать это таким образом, чтобы обрабатывать пропускную способность и быть надежным?

Для этого вы можете комбинировать конвейеры core.async и обратные вызовы http-kit. На самом деле вам не нужно создавать подпрограмму go для каждого запроса (хотя это не повредит), потому что вы можете использовать асинхронную put! из обратного вызова http-kit.

Используйте ограниченные буферы для каждого шага конвейера, чтобы ограничить количество активных подключений, которые будут (как минимум) ограничены количеством эфемерных портов TCP, доступных в вашей системе.

Вот пример небольшой программы, которая делает что-то похожее на то, что вы описали. Он считывает «события» из канала, в данном случае каждое событие имеет идентификатор «1» и ищет эти идентификаторы в службе HTTP. Он берет ответ от этого первого вызова, ищет ключ JSON "next" и ставит его в очередь в качестве URL-адреса для шага 2. Наконец, когда этот поиск завершен, он добавляет событие в канал out, который отслеживает подпрограмма go для отчета о статистике.

(ns concur-req.core
  (require [clojure.core.async :as async]
           [cheshire.core :refer [decode]]
           [org.httpkit.client :as http]))

(defn url-of
  [id]
  ;; this service responds within 100-200ms
  (str "http://localhost:28080/" id ".json"))

(defn retrieve-json-async
  [url c]
  (http/get url nil
            (fn [{body :body status :status :as resp}]
              (if (= 200 status)
                (async/put! c (decode body true))
                (println "ERROR:" resp))
              (async/close! c))))

(defn run [parallelism stop-chan]
  (let [;; allocate half of the parallelism to each step
        step1-n    (int (max (/ parallelism 2) 1))
        step2-n    step1-n
        ;; buffer to take ids, transform them into urls
        step1-chan (async/chan step1-n (map url-of))
        ;; buffer for result of pulling urls from step1, xform by extracting :next url
        step2-chan (async/chan step2-n (map :next))
        ;; buffer to count completed results
        out-chan   (async/chan 1 (map (constantly 1)))
        ;; for delivering the final result
        final-chan (async/chan)
        start-time (System/currentTimeMillis)]

    ;; process URLs from step1 and put the result in step2
    (async/pipeline-async step1-n step2-chan retrieve-json-async step1-chan)
    ;; process URLs from step2 and put the result in out
    (async/pipeline-async step2-n out-chan retrieve-json-async step2-chan)

    ;; keep the input channel full until stop-chan is closed.
    (async/go-loop []
      (let [[v c] (async/alts! [stop-chan [step1-chan "1"]])]
        (if (= c stop-chan)
          (async/close! step1-chan)
          (recur))))

    ;; count messages on out-chan until the pipeline is closed, printing
    ;; status message every second
    (async/go-loop [status-timer (async/timeout 1000) subt 0 accu 0]
      (let [[v c] (async/alts! [status-timer out-chan])]
        (cond (= c status-timer)
              (do (println subt "records...")
                  (recur (async/timeout 1000) 0 (+ subt accu)))

              (nil? v)
              (async/>! final-chan (+ subt accu))

              :else
              (recur status-timer (+ v subt) accu))))

    ;; block until done, then emit final report.
    (let [final-total (async/<!! final-chan)
          elapsed-ms  (- (System/currentTimeMillis) start-time)
          elapsed-s   (/ elapsed-ms 1000.0)]
      (print (format "Processed %d records with parallelism %d in %.3f seconds (%d/sec)\n"
                     final-total parallelism elapsed-s
                     (int (/ final-total elapsed-s)))))))

(defn run-for
  [seconds parallelism]
  (let [stop-chan (async/chan)]
    (future
      (Thread/sleep (* seconds 1000))
      (async/close! stop-chan))
    (run parallelism stop-chan)))

(do
  ;; Warm up the connection pool, avoid somaxconn problems...
  (doseq [p (map #(* 20 (inc %)) (range 25))]
    (run-for 1 p))
  (run-for (* 60 60 6) 500))

Чтобы проверить это, я настроил службу HTTP, которая отвечает только после того, как засыпает в случайное время между 100-200 мс. Затем я запускал эту программу в течение 6 часов на своем Macbook Pro.

При параллелизме, установленном на 500, я получил в среднем 1155 завершенных транзакций в секунду (2310 завершенных HTTP-запросов в секунду). Я уверен, что это могло бы быть намного выше с некоторой настройкой (и особенно с переносом службы HTTP на другую машину). Память JVM увеличилась до 1,5 ГБ в течение первых 30 минут, а затем сохранила этот размер. Я использую 64-битную JVM Oracle 1.8.

Альтернативой вашему методу A (derefing HTTP-kit, возвращенные фьючерсы внутри блока go) может быть возможность, просто делайте это таким образом, чтобы не блокировать потоки обработчика core.async в будущем, что вы можете сделать, объединив обратные вызовы httpkit и core.async:

(defn handle-event
 "Return a core.async channel that will contain the result of making both HTTP call A and B."
  [event-data]
  (let [event-a-chan (clojure.core.async/chan)
        event-b-chan (clojure.core.async/chan)
        return-chan (clojure.core.async/chan)]
    (org.httpkit.client/request "https://event-a-call"
                                {:method :get :params {"param1-k" "param1-v"}}
                                (fn [resp]
                                  (clojure.core.async/put! event-a-chan resp)))
    (org.httpkit.client/request "https://event-b-call"
                                {:method :get :params {"param1-k" "param1-v"}}
                                (fn [resp]
                                  (clojure.core.async/put! event-b-chan resp)))
    (clojure.core.async/go
      (clojure.core.async/>! return-chan {:event-a-response (clojure.core.async/<! event-a-chan)
                                          :event-b-response (clojure.core.async/<! event-b-chan)}))
    return-chan))