java - Reactor:扩展一个 ParallelFlux
问题描述
我有一组需要扩展的项目,所以我选择 reactor 是因为它的反应能力,因为扩展需要 IO 操作。
这是一段工作代码:
public Flux<Item> expand(List<Item> unprocessedItems) {
return Flux.fromIterable(unprocessedItems)
.expandDeep(this::expandItem);
}
请注意,这this::expandItem
是一个阻塞操作(多个数据库查询,一些计算,...)。现在我希望这个扩展是平行的,但据我所知.expand()
,.expandDeep()
并且只是班级的成员,Flux
而不是ParallelFlux
班级的成员。我尝试在通话之前添加.publishOn()
and ,但没有运气。.subscribeOn()
.expand()
这是我第一次使用反应器,但我没有看到任何阻止并行扩展的技术问题,有什么办法吗?API是否丢失或我错过了什么?
解决方案
是的,你是对的ParallelFlux
has not .expand()
and .expandDeep()
methods,但我可以使用其他方式,创建具有扩展方法的附加 Publisher 并将其传递给你的ParallelFlux
,如下所示:
public static void main(String[] args) {
Function<Node, Flux<Node>> expander =
node -> Flux.fromIterable(node.children);
List<Node> roots = createTestNodes();
Flux.fromIterable(roots)
.parallel(4)
.runOn(Schedulers.parallel())
.flatMap(node -> Flux.just(node).expandDeep(expander))
.doOnNext(i -> System.out.println("Time: " + System.currentTimeMillis() + " thread: " + Thread.currentThread().getName() + " value: " + i))
.sequential()
.subscribe();
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("finished");
}
我的测试数据:
static final class Node {
final String name;
final List<Node> children;
Node(String name, Node... nodes) {
this.name = name;
this.children = new ArrayList<>();
children.addAll(Arrays.asList(nodes));
}
@Override
public String toString() {
return name;
}
}
static List<Node> createTestNodes() {
return new Node("root",
new Node("1",
new Node("11")
),
new Node("2",
new Node("21"),
new Node("22",
new Node("221")
)
),
new Node("3",
new Node("31"),
new Node("32",
new Node("321")
),
new Node("33",
new Node("331"),
new Node("332",
new Node("3321")
)
)
),
new Node("4",
new Node("41"),
new Node("42",
new Node("421")
),
new Node("43",
new Node("431"),
new Node("432",
new Node("4321")
)
),
new Node("44",
new Node("441"),
new Node("442",
new Node("4421")
),
new Node("443",
new Node("4431"),
new Node("4432")
)
)
)
).children;
}
结果:
Time: 1549296674522 thread: parallel-4 value: 4
Time: 1549296674523 thread: parallel-4 value: 41
Time: 1549296674523 thread: parallel-2 value: 2
Time: 1549296674523 thread: parallel-2 value: 21
Time: 1549296674523 thread: parallel-3 value: 3
Time: 1549296674523 thread: parallel-3 value: 31
Time: 1549296674523 thread: parallel-1 value: 1
Time: 1549296674523 thread: parallel-1 value: 11
Time: 1549296674525 thread: parallel-2 value: 22
Time: 1549296674525 thread: parallel-2 value: 221
Time: 1549296674526 thread: parallel-3 value: 32
Time: 1549296674526 thread: parallel-3 value: 321
Time: 1549296674526 thread: parallel-3 value: 33
Time: 1549296674526 thread: parallel-3 value: 331
Time: 1549296674526 thread: parallel-3 value: 332
Time: 1549296674526 thread: parallel-3 value: 3321
Time: 1549296674526 thread: parallel-4 value: 42
Time: 1549296674526 thread: parallel-4 value: 421
Time: 1549296674526 thread: parallel-4 value: 43
Time: 1549296674526 thread: parallel-4 value: 431
Time: 1549296674526 thread: parallel-4 value: 432
Time: 1549296674526 thread: parallel-4 value: 4321
Time: 1549296674527 thread: parallel-4 value: 44
Time: 1549296674527 thread: parallel-4 value: 441
Time: 1549296674527 thread: parallel-4 value: 442
Time: 1549296674527 thread: parallel-4 value: 4421
Time: 1549296674528 thread: parallel-4 value: 443
Time: 1549296674528 thread: parallel-4 value: 4431
Time: 1549296674528 thread: parallel-4 value: 4432
如您所见expander
,在并行线程中工作。
推荐阅读
- c# - 如何使用 MVC Razor 在 C# MVVC 中将模型传递给视图
- django - 传递给我的 Django 视图的请求中的“购物车”cookie 是空的,尽管它显示在浏览器存储中
- php - 如何使用 TCPDF 在同一个 PDF 中正确输出多个页面?
- numpy - 使用 numpy 更新更改 numba guvectorize 输出的类类型
- typescript - 有没有办法让 npm 发布到不同于它从 package.json 推断的组织?
- python - 尝试连接散点图日期时间给我来自 matplot 的 tz 错误
- r - 如何在图表中添加黑色轮廓?
- ios - 在 UICollectionView 或 UITableView 中实现 Sticky Cell
- c# - MVC 文本框绑定多个属性
- reactjs - 是否可以使用 useSWR 在 nextJS 中发送页面 ID,以便我可以根据我所在页面的 ID 获取数据