foo.clj

;;; lazy-seq based flow

(defn proc-a [id]
  ;; 処理A
  url)
(defn proc-b [url]
  ;; 処理B
  json
  )
(defn proc-c [json]
  ;; 処理C
  )

(->> (io/reader "ids.txt")
     line-seq
     (map proc-a)
     (map proc-b)
     (map proc-c))


;;; core.async + transducer based flow
(let [in-ch (->> (io/reader "ids.txt")
                 line-seq
                 a/to-chan)
      xf (comp (map proc-a)
               (map proc-b)
               (map proc-c))
      out-ch (chan)
      worker-count 4]
  (a/pipeline worker-count out-ch xf in-ch))


;;; core.async + transducer based flow
(let [in-ch (->> (io/reader "ids.txt")
                 line-seq
                 a/to-chan)
      a-xf (map proc-a)
      b-xf (map proc-b)
      c-xf (map proc-c)
      a-worker-count 8
      a-out-ch (chan a-worker-count)
      b-worker-count 4
      b-out-ch (chan b-worker-count)
      c-worker-count 2
      c-out-ch (chan c-worker-count)]
  (a/pipeline a-worker-count a-out-ch a-xf in-ch)
  (a/pipeline b-worker-count b-out-ch b-xf a-out-ch)
  (a/pipeline-blocking c-worker-count c-out-ch c-xf b-out-ch)
  )