如何将阻塞函数的更新变量放入 ReactiveX 的异步数据流中

Question

我正在尝试发送一个 int 变量，该变量在 for 循环中不断更新。在 for 循环中，数据被发布到一个 observable。由于变量在阻塞调用中递增，我想在我的 reactiveX 订阅方法中获取这个值。注意：我不想将此变量添加到我的异步数据流中（即，不使用 publisher.onNext() 方法发送此值）。

在 for 循环的每次迭代中递增 int 变量并发布到 observable 之后，我在一个对象上调用 wait()。在订阅消费中，我取变量的值，然后 notify() 同一个对象。我收到异常“异步循环中断”

public int var=0;

main(String[] args) {
    for loop {
        variable++;
        publisher.onNext(args[0]);
        //call wait on a thread to make sure current value of 'var' 
        //is picked in the getSubscriber() method
        wait(); 
    }
}

public Subscriber<T> getSubscriber() {
    return new Subscriber<Inference>() {
            @Override public void onCompleted() {}
            @Override public void onError(Throwable e) { 
               e.printStackTrace(); 
            }
            @Override public void onNext(Inference infer) {
                //do something

                sysout(var);
                //call notify on the thread to resume control in for 
                //loop
                notify();
            }
    };
}

Answer 1

只需在synchronized 块中调用wait() 和notify() 即可。