Когда использовать неблокирующий ›! / потоки и блокировка ›!! /горутины с clojure core.async

Лично я думаю, что ваше использование потоков здесь, вероятно, правильное решение. Волшебная неблокирующая природа go-blocks происходит от «парковки», которая представляет собой особый вид псевдоблокировки, который использует конечный автомат core.async, но, поскольку ваша база данных вызывает действительно блок, а не переводит конечный автомат в припаркованное состояние. , вы просто заблокируете какой-то поток из пула потоков core.async. Это зависит от того, сколько времени занимают ваши синхронные вызовы, так что это тот случай, когда эталонные тесты могут быть информативными, но я сильно подозреваю, что потоки здесь — правильный подход.

Единственным исключением является ваша функция агрегатора. Мне кажется, что это можно просто сложить в определение чан-аута, как (def chan-out (map< aggregate chan-in)).

Для общего обзора go-blocks по сравнению с потоками Мартин Тройер написал хороший рассмотрение двух подходов и определения того, какой из них быстрее в какой ситуации. Версия Cliff Notes заключается в том, что блоки go хороши для адаптации уже асинхронных библиотек для использования с core.async, а потоки хороши для создания асинхронных процессов из синхронных частей. Например, если бы ваша база данных имела API на основе обратного вызова, то go-blocks был бы определенным выигрышем. Но так как это синхронно, они не подходят.

я думаю, что было бы лучше использовать макросы "go", чтобы иметь неблокирующие потоки в этом случае ETL.

Я написал очень простой код для достижения синхронизированной последовательности процессов, подразумеваемой в задачах извлечения, преобразования и загрузки.

Введите в ответе следующий код:

(require '[clojure.core.async :as async :refer [<! >! <!! timeout chan alt! go]])

(def output(chan))

(defn extract [origin]
  (let [value-extracted (chan)
        value-transformed (chan)
        value-loaded (chan)]
    (go
     (<! (timeout (+ 100 (* 100 (rand-int 20))))) ; wait a little
     (>! value-extracted  (str origin " > extracted  ")))
    (go
     (<! (timeout (+ 100 (* 100 (rand-int 20))))) ; wait a little
     (>! value-transformed  (str (<! value-extracted) " > transformed " )))
    (go
     (<! (timeout (+ 100 (* 100 (rand-int 20))))) ; wait a little
     (>! value-loaded  (str (<! value-transformed) " > loaded " )))
    (go
     (<! (timeout (+ 100 (* 100 (rand-int 20))))) ; wait a little
     (>! output  [origin (<! value-loaded)]))))

(go
 (loop [origins-already-loaded []]
   (let [[id message] (<! output)
         origins-updated (conj origins-already-loaded id)]
     (println message)
     (println origins-updated)
     (recur origins-updated)
     )
   ))

Введите в ответе:

(doseq [example (take 10 (range))] (extract example))

1 > extracted   > transformed  > loaded 
[1]
7 > extracted   > transformed  > loaded 
[1 7]
0 > extracted   > transformed  > loaded 
[1 7 0]
8 > extracted   > transformed  > loaded 
[1 7 0 8]
3 > extracted   > transformed  > loaded 
[1 7 0 8 3]
6 > extracted   > transformed  > loaded 
[1 7 0 8 3 6]
2 > extracted   > transformed  > loaded 
[1 7 0 8 3 6 2]
5 > extracted   > transformed  > loaded 
[1 7 0 8 3 6 2 5]
9 > extracted   > transformed  > loaded 
[1 7 0 8 3 6 2 5 9]
4 > extracted   > transformed  > loaded 
[1 7 0 8 3 6 2 5 9 4]

ОБНОВЛЕНИЕ:
исправлена ​​ошибка, заключавшаяся в использовании <!! (timeout (+ 100 (* 100 (rand-int 20))))) внутри удаленной функции "подожди-ка", которая блокировала другие процессы без блокировки.