RxJava - 为什么执行者只使用一个线程

Question

我已经创建了一个固定的线程池来处理每300毫秒发送一个事件，并假定该进程需要1000毫秒。假设多线程将工作，但只有一个线程被重用。

如果我设置的sleepTime小于300毫秒，处理线程改变，但是没用。

问题：我能做些什么来使它并发？为什么程序重用线程？

预先感谢您

public static void main(String[] args) throws InterruptedException { 
    long sleepTime = 1000; 
    ExecutorService e = Executors.newFixedThreadPool(3); 

    Observable.interval(300, TimeUnit.MILLISECONDS) 
    .subscribeOn(Schedulers.computation()) 
    .flatMap(new Func1<Long, Observable<Long>>() { 
     @Override 
     public Observable<Long> call(Long pT) { 
      return Observable.just(pT).subscribeOn(Schedulers.from(e)); 
     } 
    }) 
    .doOnNext(new Action1<Long>() { 

     @Override 
     public void call(Long pT) { 
      try { 
       Thread.sleep(sleepTime); 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } 

     } 
    }) 
    .subscribe(new Action1<Long>() { 

     @Override 
     public void call(Long pT) { 
      System.out.println("i am " + pT + "in thread:" + Thread.currentThread().getName()); 

     } 
    }); 


    Thread.sleep(50000); 
    e.shutdownNow(); 

} 

日志

i am 0in thread:pool-1-thread-1 
i am 1in thread:pool-1-thread-1 
i am 2in thread:pool-1-thread-1 
i am 3in thread:pool-1-thread-1 
i am 4in thread:pool-1-thread-1 
i am 5in thread:pool-1-thread-1 
i am 6in thread:pool-1-thread-1 
i am 7in thread:pool-1-thread-1 
i am 8in thread:pool-1-thread-1 
i am 9in thread:pool-1-thread-1 
i am 10in thread:pool-1-thread-1 
i am 11in thread:pool-1-thread-1 

Answer 1

据我了解在你的代码，生产者在比用户更快的速度生产。但是Observable<Long> interval(long interval, TimeUnit unit)实际上不支持Backpressure。文档指出

该操作员不支持背压，因为它使用时间。如果下游的 需要较慢，则应该减慢计时器或使用类似于{@link #onBackpressureDrop}的东西 。

如果你的处理真的比生产速度较慢，你可以在你的用户代码做的就是这样的事情

.subscribe(new Action1<Long>() { 

    @Override 
    public void call(Long pT) { 
     e.submit(new Runnable() { 
      System.out.println("i am " + pT + "in thread:" + Thread.currentThread().getName()); 

     } 
    } 
}); 

Answer 2

相反

.subscribeOn(Schedulers.computation()) 

尝试

.observeOn(Schedulers.computation()) 

这个我以前玩过机智的例子h与Rx的并发工作相当不错，例如

public class ObservableZip { 

private Scheduler scheduler; 
private Scheduler scheduler1; 
private Scheduler scheduler2; 

@Test 
public void testAsyncZip() { 
    scheduler = Schedulers.newThread(); 
    scheduler1 = Schedulers.newThread(); 
    scheduler2 = Schedulers.newThread(); 
    long start = System.currentTimeMillis(); 
    Observable.zip(obAsyncString(), obAsyncString1(), obAsyncString2(), (s, s2, s3) -> s.concat(s2) 
                         .concat(s3)) 
       .subscribe(result -> showResult("Async:", start, result)); 
} 




public void showResult(String transactionType, long start, String result) { 
    System.out.println(result + " " + 
           transactionType + String.valueOf(System.currentTimeMillis() - start)); 
} 


public Observable<String> obAsyncString() { 
    return Observable.just("") 
        .observeOn(scheduler) 
        .doOnNext(val -> { 
         System.out.println("Thread " + Thread.currentThread() 
                   .getName()); 
        }) 
        .map(val -> "Hello"); 
} 

public Observable<String> obAsyncString1() { 
    return Observable.just("") 
        .observeOn(scheduler1) 
        .doOnNext(val -> { 
         System.out.println("Thread " + Thread.currentThread() 
                   .getName()); 
        }) 
        .map(val -> " World"); 
} 

public Observable<String> obAsyncString2() { 
    return Observable.just("") 
        .observeOn(scheduler2) 
        .doOnNext(val -> { 
         System.out.println("Thread " + Thread.currentThread() 
                   .getName()); 
        }) 
        .map(val -> "!"); 
    } 

} 

Answer 3

我在GitHub上找到了答案！

内部observable确实发射在多线程上，但接下来做的不是。如果我想让它平行，我应该在内部可观察的情况下做。