RxJava/RxAndroid:ConnectableObservable & replay(int bufferSize)

以你之姓@ 2023-10-17 17:58 27阅读 0赞

**RxJava/RxAndroid:ConnectableObservable & replay(int bufferSize)**

import android.support.v7.app.AppCompatActivity;
    import android.os.Bundle;
    import android.util.Log;
    
    import java.util.concurrent.TimeUnit;
    
    import io.reactivex.Observable;
    import io.reactivex.observables.ConnectableObservable;
    import io.reactivex.observers.DisposableObserver;
    
    public class MainActivity extends AppCompatActivity {
        private String TAG = "输出";
    
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
    
            ConnectableObservable<Long> mConnectableObservable = Observable
                    .interval(1, TimeUnit.SECONDS)
                    .replay(3);
    
            mConnectableObservable
                    .subscribe(new DisposableObserver<Long>() {
                        @Override
                        public void onNext(Long along) {
                            Log.d(TAG, "onNext1:" + along);
                        }
    
                        @Override
                        public void onComplete() {
                            Log.d(TAG, "onComplete1");
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            Log.d(TAG, e.toString());
                        }
                    });
    
            mConnectableObservable.connect();
    
            mConnectableObservable
                    .delaySubscription(5, TimeUnit.SECONDS)
                    .subscribe(new DisposableObserver<Long>() {
                        @Override
                        public void onNext(Long along) {
                            Log.d(TAG, "onNext2:" + along);
                        }
    
                        @Override
                        public void onComplete() {
                            Log.d(TAG, "onComplete2");
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            Log.d(TAG, e.toString());
                        }
                    });
        }
    }

output:

01-26 16:05:50.755 17958-17978/zhangphil.book D/输出: onNext1:0
    01-26 16:05:51.755 17958-17978/zhangphil.book D/输出: onNext1:1
    01-26 16:05:52.755 17958-17978/zhangphil.book D/输出: onNext1:2
    01-26 16:05:53.755 17958-17978/zhangphil.book D/输出: onNext1:3
    01-26 16:05:54.755 17958-17978/zhangphil.book D/输出: onNext1:4
    01-26 16:05:54.757 17958-17979/zhangphil.book D/输出: onNext2:2
    01-26 16:05:54.757 17958-17979/zhangphil.book D/输出: onNext2:3
    01-26 16:05:54.757 17958-17979/zhangphil.book D/输出: onNext2:4
    01-26 16:05:55.755 17958-17978/zhangphil.book D/输出: onNext1:5
    01-26 16:05:55.755 17958-17978/zhangphil.book D/输出: onNext2:5
    01-26 16:05:56.755 17958-17978/zhangphil.book D/输出: onNext1:6
    01-26 16:05:56.755 17958-17978/zhangphil.book D/输出: onNext2:6
    01-26 16:05:57.755 17958-17978/zhangphil.book D/输出: onNext1:7
    01-26 16:05:57.755 17958-17978/zhangphil.book D/输出: onNext2:7
    01-26 16:05:58.755 17958-17978/zhangphil.book D/输出: onNext1:8
    01-26 16:05:58.755 17958-17978/zhangphil.book D/输出: onNext2:8
    01-26 16:05:59.755 17958-17978/zhangphil.book D/输出: onNext1:9
    01-26 16:05:59.755 17958-17978/zhangphil.book D/输出: onNext2:9
    01-26 16:06:00.755 17958-17978/zhangphil.book D/输出: onNext1:10
    01-26 16:06:00.755 17958-17978/zhangphil.book D/输出: onNext2:10
    01-26 16:06:01.755 17958-17978/zhangphil.book D/输出: onNext1:11
    01-26 16:06:01.755 17958-17978/zhangphil.book D/输出: onNext2:11
    01-26 16:06:02.755 17958-17978/zhangphil.book D/输出: onNext1:12
    01-26 16:06:02.755 17958-17978/zhangphil.book D/输出: onNext2:12
    
    ......