RxJava訂閱不工作

Question

我有下面的代碼 -

package com.test.rxjava; 

import org.reactivestreams.Subscriber; 
import org.reactivestreams.Subscription; 

import io.reactivex.Flowable; 

public class App1 { 

    public static void main(String[] args) { 

     Subscriber<Integer> subscriber = new Subscriber<Integer>() { 

      @Override 
      public void onSubscribe(Subscription s) { 
      } 

      @Override 
      public void onNext(Integer t) { 
       System.out.printf("Entry %d\n", t); 
      } 

      @Override 
      public void onError(Throwable t) { 
       System.err.printf("Failed to process: %s\n", t); 
      } 

      @Override 
      public void onComplete() { 
       System.out.println("Done"); 
      } 

     }; 
     Flowable.just(123).subscribe(subscriber); 

    } 

} 

我期待執行在onNext method.However什麼代碼happens.However如果我替換下面的代碼的最後一行，我做得到輸出。

Flowable.just(123).subscribe((t) -> System.out.println(t)); 

我不確定這裏缺少的是什麼。但是一定是。我是Rx世界的新手，可以利用一些幫助來弄清楚什麼是錯誤的。提前致謝！

Answer 1

訂閱者必須要求項目。添加s.request(Long.MAX_VALUE);在onSubscribe方法如下：

@Override 
public void onSubscribe(Subscription s) { 
    s.request(Long.MAX_VALUE); 
} 

See sample code and documentation here