使用 RxJava 的生产者-消费者 [英] Producer-consumer with RxJava
本文介绍了使用 RxJava 的生产者-消费者的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我正在尝试以 rx 样式执行一个消费者数量有限(例如 2 个)的长时间操作.
I've trying to perform a long-time operation with limited count of consumers (2 for example) in rx style.
问题是如何确保只有 2 个消费者同时执行其工作.
The problem is how to be sure that only 2 consumers perform its job at the same time.
让我们有一个消费者接口:
Lets we have a Consumer interface:
public interface Consumer{
//Take a lot of time
Observable<Result> doJob(Task task);
}
和队列类:
public class Queue {
public void enqueue(Task task){
//TODO: enqueue task and do it with limited count of Consumers
}
}
如何组织任务队列和消费者的工作?
How to organize the work of task queue and consumer?
推荐答案
使用调度程序.将并发工作限制为两个任务:
Use schedulers. To limit concurrent work to two tasks:
Scheduler scheduler = Schedulers.from(Executors.newFixedThreadPool(2));
events.flatMap(x ->
Observable.fromCallable(() -> process(x))
.subscribeOn(scheduler))
.subscribe(subscriber);
这篇关于使用 RxJava 的生产者-消费者的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文