reactor-core模块

You should never do your asynchronous work alone.

— Jon Brisbin

完成Reactor 1后写到

You should never do your asynchronous work alone.

— Stephane Maldini

完成Reactor 2后写到

首先，我们来用一个Groovy的示例来展示Core模块的功能：

  1. //初始化上下文，获取默认dispather
  2. Environment.initialize()
  4. //使用默认8192 slots的RingBufferDispatcher
  5. def dispatcher = Environment.sharedDispatcher()
  7. //创建一个回调
  8. Consumer<Integer> c = { data ->
  9.         println "some data arrived: $data"
  10. }
  12. //创建一个异常回调
  13. Consumer<Throwable> errorHandler = { it.printStackTrace }
  15. //异步调度数据
  16. dispatcher.dispatch(1234, c, errorHandler)
  18. Environment.terminate()

然后，用Reactive Streams的方式来实现

  1. //独立的异步processor
  2. def processor = RingBufferProcessor.<Integer>create()
  4. //发送数据，将会保持数据安全直到subscriber连接到processor为止
  5. processor.onNext(1234)
  6. processor.onNext(5678)
  8. //消费整型数据
  9. processor.subscribe(new Subscriber<Integer>(){
  11.   void onSubscribe(Subscription s){
  12.     //无界限的subscriber
  13.     s.request Long.MAX
  14.   }
  16.   void onNext(Integer data){
  17.     println data
  18.   }
  20.   void onError(Throwable err){
  21.     err.printStackTrace()
  22.   }
  24.   void onComplete(){
  25.     println 'done!'
  26.   }
  27. }
  29. //关闭内部线程，调用complete函数
  30. processor.onComplete()