Cách tắt tốt nhất một đường ống core.async clojure của các quy trình

Question

Tôi có ứng dụng xử lý clojure là một kênh của kênh. Mỗi bước xử lý thực hiện các phép tính của nó một cách không đồng bộ (ví dụ: tạo một yêu cầu http bằng cách sử dụng bộ http hoặc một cái gì đó), và đặt nó kết quả trên kênh đầu ra. Bằng cách này, bước tiếp theo có thể đọc từ kênh đó và thực hiện tính toán của nó.

chức năng chính của tôi trông như thế này

(defn -main [args] 
(-> file/tmp-dir 
    (schedule/scheduler) 
    (search/searcher) 
    (process/resultprocessor) 
    (buy/buyer) 
    (report/reporter))) 

Hiện nay, bước lên lịch lái các đường ống dẫn (nó đã không có một kênh đầu vào), và cung cấp các chuỗi với khối lượng công việc.

Khi tôi chạy này trong REPL:

(-main "some args") 

Nó về cơ bản chạy mãi mãi do tính vô hạn của Scheduler. Cách tốt nhất để thay đổi kiến ​​trúc này sao cho tôi có thể tắt toàn bộ hệ thống từ REPL? Việc đóng mỗi kênh có nghĩa là hệ thống chấm dứt?

Một số kênh trợ giúp có phát sóng không?

Answer 1

Bạn có thể có lịch trình của bạn alts!/alts!! trên một kênh kill và các kênh đầu vào của đường ống của bạn:

(def kill-channel (async/chan)) 

(defn scheduler [input output-ch kill-ch] 
    (loop [] 
    (let [[v p] (async/alts!! [kill-ch [out-ch (preprocess input)]] 
        :priority true)] 
     (if-not (= p kill-ch) 
     (recur)))) 

Đưa một giá trị trên kill-channel sau đó sẽ chấm dứt vòng lặp.

Về mặt kỹ thuật, bạn cũng có thể sử dụng output-ch để kiểm soát quá trình (đặt kênh đóng trả lại false), nhưng tôi thường thấy rõ ràng là xóa kênh rõ ràng, ít nhất là cho các đường ống cấp cao nhất.

Để làm cho mọi thứ trở nên thanh lịch và thuận tiện hơn (cả ở REPL và trong sản xuất), bạn có thể sử dụng Stuart Sierra's component, bắt đầu vòng lặp lên lịch (trên một sợi riêng biệt) và assoc kênh tiêu diệt thành phần của bạn phương thức start của thành phần và sau đó close! kênh tiêu diệt (và do đó chấm dứt vòng lặp) trong phương thức stop của thành phần.

Answer 2

Tôi khuyên bạn nên sử dụng một cái gì đó như https://github.com/stuartsierra/component để xử lý thiết lập hệ thống. Nó đảm bảo rằng bạn có thể dễ dàng bắt đầu và dừng hệ thống của bạn trong REPL. Sử dụng thư viện đó, bạn sẽ thiết lập nó để mỗi bước xử lý sẽ là một thành phần và mỗi thành phần sẽ xử lý việc thiết lập và tách các kênh trong các giao thức start và stop của chúng. Ngoài ra, bạn có thể có thể tạo ra một giao thức IStream cho mỗi thành phần để thực hiện và có mỗi thành phần phụ thuộc vào các thành phần thực hiện giao thức đó. Nó mua cho bạn một số mô-đun rất dễ dàng.

Bạn muốn kết thúc với một hệ thống mà trông giống như sau:

(component/system-map 
:scheduler (schedule/new-scheduler file/tmp-dir) 
:searcher (component/using (search/searcher) 
          {:in :scheduler}) 
:processor (component/using (process/resultprocessor) 
          {:in :searcher}) 
:buyer  (component/using (buy/buyer) 
          {:in :processor}) 
:report (component/using (report/reporter) 
          {:in :buyer})) 

Một điều tốt đẹp với kiểu cách tiếp cận này là bạn có thể dễ dàng thêm các thành phần nếu họ dựa trên một kênh là tốt. Ví dụ, nếu mỗi thành phần tạo ra kênh của nó bằng cách sử dụng một tap trên một nội bộ mult, bạn có thể thêm một trình ghi nhật ký cho bộ vi xử lý chỉ bằng một thành phần ghi nhật ký nhận bộ xử lý dưới dạng phụ thuộc.

:processor (component/using (process/resultprocessor) 
          {:in :searcher}) 
:processor-logger (component/using (log/logger) 
            {:in processor}) 

Tôi cũng khuyên bạn nên xem talk của mình để biết ý tưởng về cách hoạt động của nó.

Answer 3

Bạn nên xem xét sử dụng Stuart Sierra's reloaded workflow, mà phụ thuộc vào mô hình hóa các yếu tố 'đường ống' của bạn như components, như vậy bạn có thể mô hình độc thân logic của bạn là 'lớp', có nghĩa là bạn có thể kiểm soát việc xây dựng và phá hủy (start/stop) logic mỗi một trong số họ.