使用 RxJava 的生产者-消费者 [英] Producer-consumer with 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屋！