비동기 채널 버퍼 병합

Question

아마존을 검색하는 ISBN 목록이 있습니다. 나는이 문제를 순차적으로 해결 했으므로 이제는 동시성을 구현해야한다. 나는 core.async을 사용하여 그것을 시도했다. 내가 겪고있는 문제는 검색이 완료된 후에 모든 책을 컬렉션으로 병합 할 수 있어야만 도서 순위별로 정렬 할 수 있어야한다는 것입니다. 버퍼 크기가 10 인 채널 하나를 사용하고 있기 때문에이를 수행하는 방법을 잘 모릅니다. 내 접근 방식이 완전히 틀릴 수도 있습니다. 도움을 주시면 감사하겠습니다. 여기

동시성

위한 기능입니다

(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 제목 :

(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) 

채널의 모든 항목 컬렉션을 만들지 만 채널이 닫힐 때까지 reduce가 결과를 반환하지 않으므로 채널을 닫아야합니다.

Answer 2

모든 I/O 바운드 작업 (예 : 아마존 검색)을 겹치고 모든 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>))))