基于rxjava的生产消费模型

一、前言

最近在看springcloud的熔断机制的实现，发现底层使用的rxjava实现，就看了下rxjava的使用，发现rxjava使用可也便捷实现前面讲解的定时生产与消费。

二、rxjava版生产消费实现

在简单抽象下要实现的功能，定时器线程间隔3秒生成一个任务，假如任务里面有3个子任务，则消费线程要分3秒，每秒消费一个子任务。

image.png

• prouducerTimer是一个定时器做生产者使用，间隔3s调用一次run方法
• 代码（1）创建一个任务
• 代码（2）创建一个观察者对象，Observable.interval方法间隔1秒发射出一个对象，比如第一秒发射出0，第二秒发射出1....。但是我们要模拟是只需要3s的情况，所以使用take方法只获取Observable.interval观察者对象里面的TAG_QUEUE_SIZE=3个元素。这里保证了生成的观察这对象只会发射出3个元素，并且是间隔1s发出。但是这时候发射出的是0，1，2而不是代码（1）生成的任务。所以使用flatMap方法对元素进行转换，转换为代码（1）生成的元素。
• 代码（3）创建一个观察者并订阅观察者对象。

运行代码输出如下：

image.png

可知同一个元素在连续的3s内被消费了。

三、总结

rxjava的功能挺强大的，合理的排列他提供的功能可以大大简化我们的开发成本。 作者：加多