Observable.create 和 fromPublisher 有什么区别？

【问题标题】：What is the difference between Observable.create and fromPublisher?Observable.create 和 fromPublisher 有什么区别？
【发布时间】：2021-09-29 06:06:49
【问题描述】：

在下面的源代码中将Observable.create 更改为Observable.fromPublisher 不起作用。（如果 sample 不存在，则全部订阅，但如果 sample 存在，则不订阅任何内容。）

import io.reactivex.Observable;

import java.util.concurrent.TimeUnit;

public class SampleMain {
    public static void main(String[] args) {
        Observable<String> o = Observable.create(s -> {
            new Thread(() -> {
                for (int i=0; i<100; i++) {
                    s.onNext("Hello Observable.fromPublisher() A" + i);
                    s.onNext("Hello Observable.fromPublisher() B" + i);
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                s.onComplete();
            }).start();
        });
        o
                .sample(1, TimeUnit.MILLISECONDS)
                .subscribe(System.out::println);
    }
}

【问题讨论】：

标签： rx-java2

【解决方案1】：

fromPublisher 需要一个正确实现的org.reactivestreams.Publisher，它尊重rules。这些Publishers 通常来自第三方库或 API。

create 已内置基础架构，可将更简单的发射器样式 API 转换为 Observable，因此开发人员不必担心太多底层协议。

还请您注意fromPublisher 的javadoc：

Publisher 必须遵循 Reactive-Streams 规范。违反规范可能会导致未定义的行为。

如果可能，请使用 create(ObservableOnSubscribe) 创建类似源的 Observable。

请注意，尽管 Publisher 看起来是一个函数式接口，但不建议通过 lambda 实现它，因为规范要求使用无状态 lambda 无法实现的状态管理。

【讨论】：