非同期チャネルバッファをマージする

Question

Amazonに検索するISBNのリストがあります。私はすでにこれを順番に解決しているので、タスクは並行処理を実装することです。私はcore.asyncを使って試しました。私が抱えている問題は、検索が完了した後で、すべての書籍をコレクションにまとめて、ブックランクで並べ替えることができるようにする必要があることです。私はバッファサイズが10の1つのチャンネルを使用しているので、これを行う方法についてはわかりません。私のアプローチは完全に間違っているかもしれません。ヘルプは高く評価されます。

(defn get_title_and_rank_for_one_isbn [amazon_report] 
    (def book_title (get-in amazon_report [:items 0 :item-atributes :title])) 
    (def sales_rank(get-in amazon_report [:items 0 :SalesRank])) 
    (def book_isbn(get-in amazon_report [:items 0 :asin])) 
    (reduce into [[book_title] [book_isbn] [sales_rank]])) 

とコール：

(def list_of_isbns (split_isbns "src/clj_amazon/isbn_list.txt")) 
(concurrency_test list_of_isbns) 

Answer 1

あなたは

(async/reduce conj '() book-chan) 

を使用することができるはずです。ここ

は、並行

(def book_channel (chan 10)) 

(defn concurrency_test [list_of_isbns] 
     ([doseq [isbn list_of_isbns] 
     (go(>! book_channel(get_title_and_rank_for_one_isbn(amazon_search isbn))))]) 
    ) 
) 

GETタイトルの関数であります

チャンネル内のすべてのアイテムのコレクションを作成しますが、チャンネルが閉じられるまでreduceは結果を返さないため、チャンネルを閉じることを忘れないでください。

Answer 2

すべてのI/Oバウンドタスク（たとえばamazon-search）をオーバーラップし、すべてのCPUバウンドタスクを並列化する（たとえば、レポートの解析など）場合は、パイプライン関数を調べる必要があります。

(let [isbn> (chan) 
     report> (chan) 
     out> (chan)] 

    ;; pipeline-async will take isbn from isbn> channel, invoke 
    ;; amazon-search-async and pipe the result to report> channel. 

    (pipeline-async 10 ;; up to 10 I/O bound requests 
        report> 
        amazon-search-asyn 
        isbn>) 

    ;; pipeline will take report from report> channel and feed it 
    ;; to the transducer (map get-title-and-rank-etc) for processing, 
    ;; the processed report will be pushed to the out> channel. 

    (pipeline (.. Runtime getRuntime availableProcessors) 
      out> 
      (map get-title-and-rank-etc) 
      report>) 

    ;; read isbn from file and push it to isbn> channel 
    (->> "isbn_list.txt" 
     io/reader 
     line-seq 
     (onto-chan isbn>)) 

    ;; take all report from out> channel and sort it by rank & title 
    (sort-by (juxt :rank :title) (<!! (async/into [] out>))))