RxJAVA-Single

Posted 2022-11-30 征服.刘华强

Single用于只发射一个数据，并且只有一个订阅者，订阅者一共三个方法。

interface SingleObserver<T> 
    void onSubscribe(Disposable d);
    void onSuccess(T value);
    void onError(Throwable error);

Single只发射一个元素，所以没有complete 方法，不像Observable或者Flowable，数据发射完成之后，需要调用complete告诉下游已经完成。

package com.netty.demo.vertx;

import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class RxJavaTest 

    public static void main(String[] args) throws InterruptedException 

        Single.create(new SingleOnSubscribe<String>() 
            @Override
            public void subscribe(SingleEmitter<String> emitter) 
                //发送成功事件或者发送异常事件(并且事件只能发送一次)
                emitter.onSuccess("success");
                //emitter.onError(new RuntimeException("err"));
            
        ).observeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .subscribe(new SingleObserver<String>()
                    @Override
                    public void onSubscribe(Disposable d) 
                        log.info("onSubscribe");
                    
                    @Override
                    public void onSuccess(String o) 
                        log.info("onSuccess:" + o);
                    
                    @Override
                    public void onError(Throwable e) 
                        log.error("onError:" + e);
                    
                );

        Thread.sleep(300000000);