Tránh thiếu các trường hợp ngoại lệ ngoại lệ

Question

Xuất phát từ C#, khi tôi sử dụng RX và có ấn tượng, các mục liên tục sẽ được thêm vào hàng đợi nội bộ cho đến khi ứng dụng hết bộ nhớ (theo như tôi có thể gọi lại).

Trong ReactiveX (RXJava), có vẻ như chúng đã thực hiện một lập trường khác bằng cách ném ngoại lệ khi áp suất nén bắt đầu xây dựng.

Điều này có nghĩa là tôi phải sử dụng một cái gì đó như onBackpressureBuffer() và trong cuộc gọi đến subscribe() chuyển vào số Subscriber<? super T> làm cho yêu cầu tăng luồng để giải phóng áp lực.

Có thể bởi vì tôi đã sử dụng phương pháp RX.NET cho điều này, nhưng dường như nó có ý nghĩa với tôi.

Thứ nhất, tôi đã hiểu chính xác điều này chưa?

Thứ hai, có anyway tôi có thể "vô hiệu hóa" tính năng này, để nó hoạt động trong cùng một cách như RX.NET như tôi không muốn phức tạp cuộc gọi của tôi subscribe() bằng cách kiểm tra xem tôi đã thực hiện một trong những nhà khai thác áp lực này để xem liệu tôi có phải gọi số request() hay không.

Answer 1

Trong scala (Tôi không biết cú pháp Java, nhưng các cuộc gọi phương pháp sẽ giống nhau), bạn chỉ cần bật

fastHotObservable.subscribe(next => slowFunction(next))

vào

fastHotObservable.onBackpressureBuffer.subscribe(next => slowFunction(next))

Đó Hãy làm nó. Tất nhiên, khi chạy nó sẽ phải có một số giai đoạn không hoạt động, do đó, quá trình này đôi khi có thời gian để bắt kịp và xử lý các phần tử đệm.

Tôi không nghĩ rằng nó là tinh thần, tôi thấy nó tốt đẹp mà bạn có thể chọn chiến lược để xử lý unhandled backpressure mình thay vì bị buộc vào một lựa chọn cho bạn. Tôi cũng muốn phải xác định rõ ràng.

Thực tế, chiến lược mà RX.net sử dụng không phải lúc nào cũng tốt nhất. Tôi đã sử dụng một số ít nhất onBackpressureDrop cuộc gọi gần đây để quên di chuyển chuột Tôi không có thời gian để xử lý, và tôi vui vì tôi có thể tránh việc chúng được đệm dễ dàng như vậy.

Answer 2

Trường hợp ngoại lệ liên quan đến lỗi xảy ra nếu nguồn có thể quan sát không hỗ trợ ấn tượng, một vài nhà khai thác thời gian có hành vi này và nhiều hành vi quan sát được thực hiện cho phiên bản trước 0,20. Mặt khác, người đăng ký theo mặc định chạy ở chế độ không bị chặn (họ yêu cầu Long.MAX_VALUE và không bao giờ bận tâm yêu cầu thêm). Hầu hết các nhà khai thác có đường dẫn nhanh cho trường hợp này hoặc chỉ đơn giản là không can thiệp với áp suất ngược.

Nguồn phổ biến nhất của ngoại lệ là các toán tử observeOn và merge. Bạn sẽ cần phải thực hiện lại chúng. ObserveOn có thể được chuyển sang hàng đợi không bị chặn và hợp nhất có thể bỏ qua bằng cách sử dụng hàng đợi hoàn toàn. Here is an example implementation của hai toán tử này.

Answer 3

Phần lớn thời gian phương pháp tiếp cận chính xác sẽ là sử dụng chiến lược áp lực trở lại tốt như onBackPressureDrop. Tuy nhiên, ghi nhớ để làm điều này là dễ bị lỗi.

Khi chúng tôi chuyển đổi sang RxJava2, chúng tôi nhận thấy rằng khoảng 100% các Đài quan sát của chúng tôi chỉ muốn sử dụng onBackPressureBuffer.Với điều này là trường hợp chúng tôi tăng kích thước rx.ring-buffer.size, ngăn chặn tất cả các va chạm áp lực trở lại và chuyển sang với cuộc sống của chúng tôi.

https://eng.uber.com/rxjava-backpressure/