Chia Rx Quan sát được thành nhiều luồng và xử lý riêng lẻ

Question

Đây là hình ảnh về những gì tôi đang cố gắng hoàn thành.

--abca - bbb - một sự chia rẽ

vào

--một ----- một ------- a -> một dòng

- --- b ------ bbb --- -> dòng b

------ c ---------- -> dòng c

Sau đó, , có thể

a.subscribe() 
b.subscribe() 
c.subscribe() 

Cho đến nay, mọi thứ tôi đã tìm thấy đã tách luồng bằng cách sử dụng một nhómBởi(), nhưng sau đó thu gọn mọi thứ lại thành một luồng và xử lý tất cả chúng trong cùng một hàm. Những gì tôi muốn làm là xử lý từng luồng có nguồn gốc theo một cách khác.

Cách tôi đang thực hiện ngay bây giờ là thực hiện một loạt các bộ lọc. Có cách nào tốt hơn để làm điều này?

Answer 1

Bạn không phải thu gọn Observables từ groupBy. Thay vào đó, bạn có thể đăng ký với họ.

Something như thế này:

String[] inputs= {"a", "b", "c", "a", "b", "b", "b", "a"}; 

    Action1<String> a = s -> System.out.print("-a-"); 

    Action1<String> b = s -> System.out.print("-b-"); 

    Action1<String> c = s -> System.out.print("-c-"); 

    Observable 
      .from(inputs) 
      .groupBy(s -> s) 
      .subscribe((g) -> { 
       if ("a".equals(g.getKey())) { 
        g.subscribe(a); 
       } 

       if ("b".equals(g.getKey())) { 
        g.subscribe(b); 
       } 

       if ("c".equals(g.getKey())) { 
        g.subscribe(c); 
       } 
      }); 

Nếu báo cáo trông kinda xấu xí nhưng ít nhất bạn có thể xử lý từng dòng riêng biệt. Có thể có cách tránh chúng.

Answer 2

dễ dàng như chiếc bánh, chỉ cần sử dụng filter

Một ví dụ trong scala

import rx.lang.scala.Observable 

val o: Observable[String] = Observable.just("a", "b", "c", "a", "b", "b", "b", "a") 
val hotO: Observable[String] = o.share 
val aSource: Observable[String] = hotO.filter(x ⇒ x == "a") 
val bSource: Observable[String] = hotO.filter(x ⇒ x == "b") 
val cSource: Observable[String] = hotO.filter(x ⇒ x == "c") 

aSource.subscribe(o ⇒ println("A: " + o), println,() ⇒ println("A Completed")) 

bSource.subscribe(o ⇒ println("B: " + o), println,() ⇒ println("B Completed")) 

cSource.subscribe(o ⇒ println("C: " + o), println,() ⇒ println("C Completed")) 

Bạn chỉ cần chắc chắn rằng nguồn quan sát được là nóng. Cách dễ nhất là share.