这是我第一次尝试 RxJava。
我正在尝试用于一项简单的任务:需要通过 Java 应用程序从数据库导出数据。导出分三步完成:
- 发出查询以查找需要导出的对象的主键。
- 批处理这些 ID，并并行获取/编码完整对象。
- 将编码对象写入输出流。
我认为一个好的方法是让一个线程执行步骤 1(迭代 ResultSet 的页面)，让一组线程执行步骤 2(每个 ResultSet 页面一个任务)，并让主线程执行步骤 3(这需要在单个线程上发生)。
我知道在 Rx 世界中阻塞主线程并不被视为一件好事，但对于这个问题，让我们忽略这一点。这是我第一次在遗留应用程序上引入响应式(Reactive)编程。
上述场景的测试如下所示:
@Test
public void testSimplePipelineRx()
{
Scheduler idsScheduler = makeScheduler("idsExecutor");
Scheduler dataScheduler = makeScheduler("dataExecutor");
final List<MutablePair<Integer, List<Integer>>> stateHolder = new CopyOnWriteArrayList<>();
Observable<Integer> idsObservable = Observable.create(SyncOnSubscribe.createSingleState(
() ->
{
print("ids observable initialized");
stateHolder.add(emitIds());
return stateHolder.get(0);
},
(state, observer) ->
{
if (state.getLeft() >= state.getRight().size())
{
print("ids observable next - emitting onComplete");
observer.onCompleted();
}
else
{
Integer val = state.getRight().get(state.getLeft());
state.setLeft(state.getLeft() + 1);
print("ids observable next - emitting " + val);
observer.onNext(val);
}
},
(state) ->
{
print("ids observable finish");
state.setLeft(-1);
state.getRight().clear();
}
));
final ConcurrentHashMap<String, Boolean> results = new ConcurrentHashMap<>();
print("Starting");
idsObservable
.buffer(2)
.flatMap(i -> Observable.just(i)
.observeOn(dataScheduler)
.map(k ->
{
print("Transforming values: " + k.get(0) + "-" + k.get(k.size() - 1));
return "Values: " + k.get(0) + "-" + k.get(k.size() - 1);
})
, 5 //max count flatMap will have queued up
)
.subscribeOn(idsScheduler)
.toBlocking()
.subscribe(new Subscriber<String>()
{
@Override
public void onStart()
{
request(5);
}
@Override
public void onCompleted()
{
print("Observed done");
}
@Override
public void onError(Throwable err)
{
print(ExceptionUtils.getStackTrace(err));
print("Observed error");
}
@Override
public void onNext(String str)
{
print("Observed value " + str);
results.put(str, true);
request(1);
}
});
print("Asserting");
Assert.assertEquals(Integer.valueOf(-1), stateHolder.get(0).getLeft());
Assert.assertEquals(0, stateHolder.get(0).getRight().size());
Assert.assertEquals(7, results.keySet().size());
print("Finishing");
}
private MutablePair<Integer, List<Integer>> emitIds()
{
return new MutablePair<>(0, IntStream.range(0, 13).mapToObj(i -> i).collect(Collectors.toList()));
}
private Scheduler makeScheduler(String name)
{
ExecutorService executor = Executors.newFixedThreadPool(10, new ThreadFactory()
{
private AtomicInteger id = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r)
{
return new Thread(r, name + "-" + id.getAndIncrement());
}
});
return Schedulers.from(executor);
}
private void print(String msg)
{
System.out.println(new SimpleDateFormat("HH:mm:ss.SSS").format(new Date()) + " - " + Thread.currentThread().getName() + " - " + msg);
}
我得到的输出是这样的:
16:25:36.168 - main - Starting
16:25:36.185 - idsExecutor-0 - ids observable initialized
16:25:36.194 - idsExecutor-0 - ids observable next - emitting 0
16:25:36.194 - idsExecutor-0 - ids observable next - emitting 1
16:25:36.201 - idsExecutor-0 - ids observable next - emitting 2
16:25:36.202 - dataExecutor-0 - Transforming values: 0-1
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 3
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 4
16:25:36.202 - dataExecutor-1 - Transforming values: 2-3
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 5
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 6
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 7
16:25:36.203 - idsExecutor-0 - ids observable next - emitting 8
16:25:36.203 - idsExecutor-0 - ids observable next - emitting 9
16:25:36.203 - dataExecutor-3 - Transforming values: 6-7
16:25:36.203 - dataExecutor-2 - Transforming values: 4-5
16:25:36.204 - dataExecutor-4 - Transforming values: 8-9
16:25:36.206 - main - Observed value Values: 0-1
****16:25:36.206 - dataExecutor-3 - ids observable next - emitting 10
****16:25:36.206 - dataExecutor-3 - ids observable next - emitting 11
16:25:36.206 - main - Observed value Values: 2-3
****16:25:36.206 - dataExecutor-3 - ids observable next - emitting 12
16:25:36.206 - dataExecutor-5 - Transforming values: 10-11
16:25:36.206 - main - Observed value Values: 4-5
****16:25:36.206 - dataExecutor-3 - ids observable next - emitting onComplete
16:25:36.207 - main - Observed value Values: 6-7
****16:25:36.207 - dataExecutor-3 - ids observable finish
16:25:36.207 - main - Observed value Values: 8-9
16:25:36.207 - dataExecutor-6 - Transforming values: 12-12
16:25:36.207 - main - Observed value Values: 10-11
16:25:36.208 - main - Observed value Values: 12-12
16:25:36.208 - main - Observed done
16:25:36.208 - main - Asserting
16:25:36.208 - main - Finishing
为什么“dataExecutor”线程在最后接管“发射”值？我本以为“idsExecutor-0”线程是唯一一个“发出”值的线程。
最佳答案
这是 RxJava 和 observeOn 中基于协同例程的背压设计的效果。运算符:来自 observeOn 的发出线程的请求调用可以执行 SyncOnSubscribe 内的生成器(这称为弱流水线)。确保SyncOnSubscribe从已知线程生成项目，使用 subscribeOn紧接着它(这称为强管道):
Observable.range(1, 5)
.subscribeOn(Schedulers.io())
.map(v -> Thread.currentThread() + "|" + v)
.observeOn(Schedulers.single())
.subscribe(w -> Thread.currentThread() + "||" + w);
Thread.sleep(1000);
关于java - RxJava subscribeOn 和observeOn 以及阻塞w/flatMap。谁做了什么？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/43879342/