core.async+transducers Shibuya.lisp #21
3 likes1,468 views
Presentation at Shibuya.lisp #21 Brief introduction on transducers + core.async. Mostly Japanese
![lazy-seqベースのフロー
;;; 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))](https://image.slidesharecdn.com/core-141029202931-conversion-gate02/85/core-async-transducers-Shibuya-lisp-21-3-320.jpg)
![core.async + transducer
;;; 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))
複数のスレッドで処理](https://image.slidesharecdn.com/core-141029202931-conversion-gate02/85/core-async-transducers-Shibuya-lisp-21-5-320.jpg)
![core.async + transducer 2
;;; 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)
)
処理ごとにワーカ数を指定可能](https://image.slidesharecdn.com/core-141029202931-conversion-gate02/85/core-async-transducers-Shibuya-lisp-21-6-320.jpg)
core.async+transducers Shibuya.lisp #21
- 1. Core.async & transducers @ktsujister
- 2. こんな処理があるとします ? input: file ? 一行ごとにIDが記載されている ? 処理A ? 受け取ったIDを元に、URLを作成 ? 処理B ? 受け取ったURLを元に、外部サイトからjsonをHTTP GET ? 処理C ? 受け取ったjsonをパースして出力ファイルに書き込む
- 3. lazy-seqベースのフロー ;;; 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))
- 4. 弱点 ? cpu-coreが複数あっても、一個しか忙しくない ? http getを待っている間に処理が行われない
- 5. core.async + transducer ;;; 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)) 複数のスレッドで処理
- 6. core.async + transducer 2 ;;; 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) ) 処理ごとにワーカ数を指定可能
- 7. transducer ? http://clojure.org/transducers ? clojure 1.7以降 ? map系変換処理をcomposeできる ? transducerが作れるfunction ? map cat mapcat filter remove take take-while take-nth drop drop-while replace partition-by partition-all keep keep-indexed dedupe random-sample ? core.async supports transducers! ? clojure.coreのみで使用する場合は、transduceで実行
- 9. References ? https://gist.github.com/ktsujister/ 67b68c059e8b6b540cf3 ? http://clojure.org/transducers ? https://www.youtube.com/watch?v=6mTbuzafcII