Rxjava 中的条件多播

Question

有无限的消息流发送到 Dispatch（实现为 observable）。 Dispatch 组件需要将消息多播给订阅者（S1、S2、S3、S4）。

• subscriber1 应该只接收 A、B 和 C 类型的消息。
• subscriber2 应该只接收 B 类型的消息
• subscriber3 应该只接收 C 类型的消息
• subscriber4 应该只接收 B 和 C 类型的消息

我是 RxJava 新手，对此推荐的模式是什么？创建 ConnectableObservable 并在订阅时进行过滤？使用主题？

Answer 1

您可以使用filter() 方法只接收您感兴趣的对象：

PublishSubject<Object> subject = PublishSubject.create();
    
Disposable sub1 = subject
    .filter(it -> it instanceof A || it instanceof B || it instanceof C)
    .subscribe(it -> {
        System.out.println("From A,B or C: "+it);
    });     
Disposable sub2 = subject
        .filter(it -> it instanceof B)
        .subscribe(it -> {
            System.out.println("From only B: "+it);
        });
Disposable sub3 = subject
        .filter(it -> it instanceof C)
        .subscribe(it -> {
            System.out.println("From only C: "+it);
        });
Disposable sub4 = subject
        .filter(it -> it instanceof B || it instanceof C)
        .subscribe(it -> {
            System.out.println("From B or C: "+it);
        });

System.out.println("-- start --");
subject.onNext(new A());
System.out.println("-- 1 --");
subject.onNext(new B());
System.out.println("-- 2 --");
subject.onNext(new C());
System.out.println("-- 3 --");
subject.onNext(new A());
System.out.println("-- 4 --");
subject.onNext(new C());
System.out.println("-- end --");

此示例代码将生成以下示例输出：

-- start --
From A,B or C: testing.A@57fa26b7
-- 1 --
From A,B or C: testing.B@5f8ed237
From only B: testing.B@5f8ed237
From B or C: testing.B@5f8ed237
-- 2 --
From A,B or C: testing.C@2f410acf
From only C: testing.C@2f410acf
From B or C: testing.C@2f410acf
-- 3 --
From A,B or C: testing.A@47089e5f
-- 4 --
From A,B or C: testing.C@4141d797
From only C: testing.C@4141d797
From B or C: testing.C@4141d797
-- end --