RxJAVA 如何动态接收数据？

Question

最近在看RXJAVA，有一个问题没有理解，就是在创建了Observable后，好像只能消费一次，但是我认为是可以消费多次的；

@Test
    public void helloworld(){
        Observable<Integer> observable = Observable.create(e -> {
            e.onNext(1);
            e.onNext(2);
            e.onNext(3);
            e.onComplete();
        });
        Observer<Integer> observer= new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("建立订阅关系");
            }
            @Override
            public void onNext(Integer value) {
                System.out.println("接收到数据：" + value);
                if(value == 3){
                    int a = 1 / 0;
                }
            }
            @Override
            public void onError(Throwable e) {
                System.out.println("出现异常：" + e.getMessage());
            }
            @Override
            public void onComplete() {
                System.out.println("完成");
            }
        };
        observable.subscribe(observer);
        
        //如何让observable中动态添加数据，然后observer再次执行onNext方法呢？？？
    }