何时使用非阻塞>!/线程和阻止> !!/goroutines with clojure core.async

我正在编写一个ETL过程来从产品数据库中读取事件级数据,转换/聚合它并写入分析数据仓库.我正在使用clojure的core.async库将这些进程分成并发执行的组件.这是我的代码的主要部分现在看起来像

    (ns data-staging.main
        (:require [clojure.core.async :as async])
        (:use [clojure.core.match :only (match)]
              [data-staging.map-vecs]
              [data-staging.tables])
        (:gen-class))

    (def submissions (make-table "Submission" "Valid"))
    (def photos (make-table "Photo"))
    (def videos (make-table "Video"))
    (def votes (make-table "Votes"))

    ;; define channels used for sequential data processing
    (def chan-in (async/chan 100))
    (def chan-out (async/chan 100))

    (defn write-thread [table]
        "infinitely loops between reading subsequent 10000 rows from 
         table and ouputting a vector of the rows(maps) 
         into 'chan-in'"
        (while true
            (let [next-rows (get-rows table)]
                (async/>!! chan-in next-rows)
                (set-max table (:max-id (last next-rows))))))

    (defn aggregator []
        "takes output from 'chan-in' and aggregates it by coupon_id, date.
         then adds / drops any fields that are needed / not needed and inputs
         into 'chan-out'"
        (while true
            (->>
                (async/!! chan-out))))

    (defn read-thread []
        "reads data from chan out and interts into Analytics DB" 
        (while true 
            (upsert (async/

正如你所看到的,我将每个os组件放到它自己的线程上并使用阻塞> !! 打电话给频道.感觉就像使用非阻塞>!对于此用例,调用以及go例程可能更好,特别是对于数据库读取,这些读取花费大部分时间执行i/o并等待产品db中的新行.是这种情况,如果是这样,实施它的最佳方法是什么？我对这两种方法之间的所有权衡以及如何有效地使用go例程有点不清楚.此外,任何其他关于如何改善整体架构的建议将非常感谢!