Rxjava并行执行耗时操作使用zip和merge

Posted 2023-02-11 zhang106209

Rxjava使用的是zip()和merge()两个方法来实现并发操作。

zip()的方法使用如下：

/**
     * zip()可以并发
     * Observable..subscribeOn(Schedulers.newThread())
     * 一起执行完,再返回结果
     * 适合处理一起展示的场景,类似Countdownlatch
     */
    public void zipData() 
        List<Observable<Integer>> observables = initObservables();
        Observable.zip(observables.get(0), observables.get(1)
                , Integer::sum)
                .subscribeOn(Schedulers.newThread())
                .observeOn(androidSchedulers.mainThread())
                .subscribe(new Observer<Integer>() 
                    @Override
                    public void onSubscribe(Disposable d) 

                    

                    @Override
                    public void onNext(Integer o) 
                        Log.e("--zw--", "zipData----onNext" + o);
                    

                    @Override
                    public void onError(Throwable e) 

                    

                    @Override
                    public void onComplete() 

                    
                );
    

zip()方法执行结果：

merge()的方法使用如下：

 public void mergeData() 
        List<Observable<Integer>> observables = initObservables();
        Observable.merge(observables)
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Integer>() 
                    @Override
                    public void onSubscribe(Disposable d) 

                    

                    @Override
                    public void onNext(Integer o) 
                        LogUtils.e("--zw--", "mergeData---onNext" + o);
                    

                    @Override
                    public void onError(Throwable e) 

                    

                    @Override
                    public void onComplete() 

                    
                );
    

merge()方法执行结果：

具体的类如下：

package com.zw.baselib.utils;

import android.util.Log;


import java.util.ArrayList;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;


/**
 * 描述:
 *
 * @author: zw
 * @time: 2022/11/18 17:15
 */
public class RxUtils 


    /**
     * 并行执行
     * merge()可以并发
     * Observable.subscribeOn(Schedulers.newThread())
     * 谁先执行完,先返回谁的结果
     */
    public void mergeData() 
        List<Observable<Integer>> observables = initObservables();
        Observable.merge(observables)
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Integer>() 
                    @Override
                    public void onSubscribe(Disposable d) 

                    

                    @Override
                    public void onNext(Integer o) 
                        LogUtils.e("--zw--", "mergeData---onNext" + o);
                    

                    @Override
                    public void onError(Throwable e) 

                    

                    @Override
                    public void onComplete() 

                    
                );
    

    /**
     * zip()可以并发
     * Observable..subscribeOn(Schedulers.newThread())
     * 一起执行完,再返回结果
     * 适合处理一起展示的场景,类似Countdownlatch
     */
    public void zipData() 
        List<Observable<Integer>> observables = initObservables();
        Observable.zip(observables.get(0), observables.get(1)
                , Integer::sum)
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Integer>() 
                    @Override
                    public void onSubscribe(Disposable d) 

                    

                    @Override
                    public void onNext(Integer o) 
                        Log.e("--zw--", "zipData----onNext" + o);
                    

                    @Override
                    public void onError(Throwable e) 

                    

                    @Override
                    public void onComplete() 

                    
                );
    

    /**
     * 创建写入的两个Observable
     *
     * @return
     */
    private List<Observable<Integer>> initObservables() 
        List<Observable<Integer>> obsList = new ArrayList<>();
        Observable<Integer> obs1 = observable(1);
        Observable<Integer> obs2 = observable(2);
        obsList.add(obs1);
        obsList.add(obs2);
        return obsList;
    


    private <T> Observable<T> observable(T str) 
        return Observable.create((ObservableOnSubscribe<T>) emitter -> 
            Log.e("--zw--", str + "---" + Thread.currentThread().getName());
            emitter.onNext(str);
            emitter.onComplete();
        ).subscribeOn(Schedulers.newThread());
    

参考文章地址