組合就是多輸入，單輸出，大家看具體的例子

package com.example.acer.rxjavatest;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;

import rx.Observable;
import rx.Observer;

//需求，列印一組使用者名稱
public class MainActivity extends AppCompatActivity {



    @Override
    protected
 
 void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
         String[] args1=new String[]{"張欣1","張欣2","張欣3","張欣4"};
        String[] args2=new String[]{"春曉1","春曉2","春曉3","春曉4"};
        //相同的陣列可以進行合併
        Observable<String> mergeObservable=Observable.merge(Observable.from(args1),Observable.from(args2));
        mergeObservable.subscribe(new
 
 Observer<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {
                Log.i("zhang_xin",s);
            }
        });
    }



    }

 

看下執行效果

注意：在上面的程式中我們還可以把merge換成mergeDelayError();

 Observable<String> mergeObservable=Observable.mergeDelayError(Observable.from(args1),Observable.from(args2));

mergeDelayError()方法:在Observable中，一旦某一個時間丟擲異常，後面的序列將會終止，如果我們希望在序列出錯的時候不影響後面的序列，那麼可以使用mergeDelayErroe()方法！
執行結果同merge同，這裡就不展示了！
merge是把兩個陣列 合併為一個數組，新的陣列長度是原來兩個陣列長度之和；

zip合併後的陣列長度是兩個陣列長度最小的那一個數組的長度；而且我們可以自己定義合併的方式，即合併的演算法自己決定：看程式碼

package com.example.acer.rxjavatest;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;

import rx.Observable;
import rx.Observer;
import rx.functions.Func2;

//需求，列印一組使用者名稱
public class MainActivity extends AppCompatActivity {



    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        //陣列長度5
         String[] args1=new String[]{"張欣1","張欣2","張欣3","張欣4","zhangxin5"};
         //陣列長度4
        String[] args2=new String[]{"春曉1","春曉2","春曉3","春曉4"};
        //相同的陣列可以進行合併
        Observable<String> mergeObservable=Observable.zip(Observable.from(args1), Observable.from(args2), new Func2<String, String, String>() {
            @Override
            public String call(String s, String s2) {
                //這裡使我們合併陣列的演算法
                return s+"------"+s2;
            }
        });
        mergeObservable.subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {
                Log.i("zhang_xin",s);
            }
        });
    }



    }

看下join方法，陣列長度是兩個陣列的乘積；

package com.example.acer.rxjavatest;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Observer;
import rx.functions.Func1;
import rx.functions.Func2;

//需求，列印一組使用者名稱
public class MainActivity extends AppCompatActivity {


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        String[] args1 = new String[]{"張欣1", "張欣2", "張欣3", "張欣4", "zhangxin5"};
        String[] args2 = new String[]{"春曉1", "春曉2", "春曉3", "春曉4"};
        Integer[] args3 = new Integer[]{1, 2, 3, 4, 5, 6};
        Observable<String> o1 = Observable.from(args1);
        Observable<String> o2 = Observable.from(args2);
        //相同的陣列可以進行合併
        Observable<String> joinObservable = o2.join(o1, new Func1<String, Observable<Long>>() {
            @Override
            public Observable<Long> call(String s) {
                return Observable.timer(2, TimeUnit.SECONDS);
            }
        }, new Func1<String, Observable<Long>>() {
            @Override
            public Observable<Long> call(String s) {
                return Observable.timer(2, TimeUnit.SECONDS);
            }
        }, new Func2<String, String, String>() {
            @Override
            public String call(String s, String s2) {
                return s + "-&--" + s2;
            }
        });
        joinObservable.subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {
                Log.i("zhang_xin", s);
            }
        });
    }


}