(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-in)
                aggregate
                (async/>!! chan-out))))
 
    (defn read-thread []
        "reads data from chan out and interts into Analytics DB" 
        (while true 
            (upsert (async/<!! chan-out))))
 
    (defn -main []
        (async/thread (write-thread submissions))
        (async/thread (write-thread photos))
        (async/thread (write-thread videos))
        (async/thread-call aggregator)
        (async/thread-call read-thread))

你可以看到,我将每个os组件放在自己的线程上并使用阻止>！呼叫频道.感觉就像使用非阻塞>！对于这种用例,调用以及执行例程可能更好,特别是对于大部分时间执行i / o并等待产品数据库中的新行的数据库读取.是这样的,如果是这样,实施它的最好办法是什么？我对两种方法之间的所有权衡有一点不清楚,以及如何有效地使用go例程.还有关于如何提高整体架构的任何其他建议将非常感谢！