RxJava2合并和多播事件流

Observable.merge(Observable.just(1, 2, 3), Observable.just(4, 5, 6))
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
        }
    });

ConnectableObservable<Integer> source = Observable.just(1, 2, 3).publish();
source.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        System.out.println("Observer 1: " + integer);
    }
});
source.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        System.out.println("Observer 2: " + integer);
    }
});
source.connect();