RXJava - 线程优先级

Question

我正在使用以下内容为我的订阅获取ExecutorService：

public static ExecutorService getThreadPoolExecutorService(int threads)
{
    int NUMBER_OF_CORES = Runtime.getRuntime().availableProcessors();
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            NUMBER_OF_CORES * 2,
            NUMBER_OF_CORES * 2,
            60L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>()
    );
    return Executors.newFixedThreadPool(threads, threadPoolExecutor.getThreadFactory());
}

并像下面这样使用它：

public static Scheduler getBackgroundSchedular()
{
    if (mBackgroundExecutorService == null)
        mBackgroundExecutorService = ThreadUtil.getThreadPoolExecutorService(4);
    return Schedulers.from(mBackgroundExecutorService);
}

我将这个时间表用于应该在后台运行的 observables。

问题

如何为 RXJava 使用PriorityBlockingQueue？通常我会使用一些特殊的runnables来实现比较函数，并为PriorityBlockingQueue使用相应的比较函数，并将上面示例中的LinkedBlockingQueue替换为PriorityBlockingQueue，但是如何使用RXJava observables做到这一点？

Answer 1

RxJava 中的调度程序与数据流正交，根本不了解它们。他们所做的只是执行一个 Runnable 实例，并且由操作员创建的所有 Runnable 生来都是平等的。因此，将优先级队列与调度程序一起使用是没有意义的。

此外，数据流是顺序的，除非存在线程交换之类的边界，否则它们会保持顺序。除非您收集事件并手动进行排序，否则具有优先级队列的操作员很有可能会按照您的预期重新排序数据。

编辑

有点不合常规，但您也可以将 PriorityBlockingQueue 放置在主题和一些运算符之间，这样您就可以为序列提供任务：

PriorityBlockingQueue<Task> q = ...

Subject<Integer, Integer> subject = PublishSubject.<Integer>create().toSerialized();

subject
.map(v -> q.poll())
.doOnNext(v -> v.execute())
.subscribe();

q.offer(new Task(...));
subject.onNext(1);