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