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); 
} 

И я использую этот Schedular для моих наблюдаемых, что должна бежать в фоновом режиме.

Вопрос

Как использовать PriorityBlockingQueue для RXJava? Обычно я использую некоторые специальные runnables, которые реализуют функцию сравнения и используют соответствующую функцию сравнения для PriorityBlockingQueue и заменяют LinkedBlockingQueue в примере выше на PriorityBlockingQueue, но как это сделать с помощью RXJava-наблюдаемых?

Answer 1

Планировщики в RxJava являются ортогональными к потоку данных и не знают о них вообще. Все, что они делают, это выполнить экземпляр Runnable, и все Runnables, созданные операторами, рождаются равными. Таким образом, не имеет смысла использовать очереди приоритетов с Schedulers.

Кроме того, потоки данных последовательны и, если нет какой-либо границы, такой как обмен нитями, они сохраняют свой порядок. Если вы не соберете события и не сделаете заказ вручную, есть вероятность, что оператор с приоритетной очередью будет переупорядочивать данные, как вы ожидаете.

Редактировать

Немного нетрадиционный, но вы могли бы также заложить 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);