Java响应式编程之handle用法解析

@GetMapping("/a")
    public Mono<Object> a() throws InterruptedException {
        long begin = System.currentTimeMillis();
        Mono<String> m5 = Mono.just("").map(s -> {
            try {
                TimeUnit.SECONDS.sleep(5);
                System.out.println("5");
                System.out.println(System.currentTimeMillis() - begin);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "5";
        }).subscribeOn(Schedulers.boundedElastic());

        Mono<String> m8 = Mono.just("").map(s -> {
            try {
                TimeUnit.SECONDS.sleep(8);
                System.out.println("8");
                System.out.println(System.currentTimeMillis() - begin);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "8";
        }).subscribeOn(Schedulers.boundedElastic());
        Mono<String> m10 = Mono.just("").map(s -> {
            try {
                TimeUnit.SECONDS.sleep(15);
                System.out.println("15");
                System.out.println(System.currentTimeMillis() - begin);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return "15";
        }).subscribeOn(Schedulers.boundedElastic());
        
        Mono<Object> single = Flux.merge(m8, m5, m10).handle(((s, synchronousSink) -> {
            if ("8".equals(s)) {
                synchronousSink.next("卢本伟牛");
                synchronousSink.complete();
            }
        })).defaultIfEmpty("null").single();
        return single;
    }

//这个s代表前面传入的元素 执行完会传入这个handle  
// synchronousSink.next() 代表我执行完这个流返回给下一个进行处理  next(Object)  
// synchronousSink.complete(); 代表我们执行完取消其他的流
Flux.merge(m8, m5, m10).handle(((s, synchronousSink) -> {

            if ("8".equals(s)) {
                synchronousSink.next("卢本伟牛");
                synchronousSink.complete();
            }
        })).subscribe(s -> {
                    System.err.println(s);
                });

public class UIHandler<T> extends Handler{
 
     protected WeakReference<T> ref;
 
     public UIHandler(T cla){
        ref = new WeakReference<T>(cla);
     }
 
      public T getRef(){
          return ref != null ? ref.get() : null;
      }
}

public class MainActivity extends Activity｛
 
    private final MainHandler mHandler = new MainHandler(this);
 
    @Override
    protected void onCreate(Bundle savedInstanceState){
        super.oncreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        mHandler.post(mRunnable);
    }
 
    private static final Runnable mRunnable = new Runnable(){
 
        @Override
         public void run(){
 
         }
    };
 
    private class MainHandler extends UIHandler{
         private MainHandler(MainActivity activity){
             super(activity);
         }
         
         @Override
         public void handleMessage(Message msg){
             super.handleMessage(msg);
             MainActivity activity = (MainActivity)ref.get();
             if(activity != null){
                if (activity.isFinishing()
                    return;
　　　　　　　　　　switch(msg.what){
　　　　　　　　　　　　case 1:
　　　　　　　　　　　　break;
　　　　　　　　　　}
             }
         }
    }
｝