问题标签 [java-flow]

我一直在玩 Java Flowoffer运算符，但是在阅读了文档并进行了测试后，我不明白。

这是我的测试

报价运算符接收要发出的项目和 BiPredicate 函数，据我了解阅读文档，只有在谓词函数为真的情况下才会发出项目。

Bur通过测试后的结果是

如果我返回 true 而不是 false，则结果没有变化。

任何人都可以更好地解释我这个操作员。

我开始使用 Java 9 Flow API，我发现并且不喜欢的第一件事是，当我们将订阅者实现传递给发布者时，我们似乎不能使用 lambda，就像我们可以使用 RxJava 一样

所以我必须定义和实现我自己的订阅者类

然后把它传给我的出版商

这真的很冗长，据我了解，这是因为我们需要在onSubscribe回调 中设置订阅

以后用于onNext继续排放subscription.request(1);

我仍然不明白为什么需要这种机制，但它避免了像我们在 RxJava 中那样使用 Lambda 作为这个例子

我想这是不可能的，我不会错过任何东西，对吧？

我再次将 RxJava 与 Java 9 Flow 进行比较。我看到 Flow 默认情况下是异步的，我想知道是否有办法让它同步运行。

有时我们只是想将它不是用于 Nio 而是用于糖语法，并拥有更同质的代码。

在 RxJava 中，默认情况下它是同步的，您可以使用observerOn和subscribeOn在您的管道中使其异步运行。

Flow中是否有任何运算符可以使其在主线程中运行？

问候。

最近发现projectreactor.io对Publisher的支持不错：

对处理器有什么好的支持吗？ 我的意思是类似或类似的东西：

如果没有，我该如何实现自己的，或者为什么我不能这样做？

更新1：

经过讨论（见评论），似乎在我的用例中我需要使用flatMap（见答案），我的问题是处理器的良好实现，我的意思是一些功能，如果它失败了，我可以控制并发出错误。我认为flatMap会给你足够的功能。就我而言，我使用了：

我正在使用 JDK 9 Flow API 创建一个用户事件系统，所以我有一个房间（它扩展了UserSubscriver上面的类），它可能有很多用户，每个用户都可以随时提供（调度）更新。

用户等级：

在onUpdate我可以接收任何参数，所以我可以接收更新的发布者，但是在onComplete.

我如何知道谁是onComplete活动的发布者？

我正在使用 JDK 9 Flow API 创建一个用户事件系统，所以我有一个房间（它实现了Flow.Subscriber<Notification>），它可能有很多用户，每个用户都可以随时提供（调度）更新。

当用户进入房间时，我订阅房间的更新user.subscribe(this)。但是没有退订，用户离开房间时如何退订？

用户等级：

我正在尝试实现 java 反应流（观察者设计模式）。

我利用了可用的样本。我向订阅者发送值没有问题，但我不能让订阅者向发布者发送一个值。有什么办法可以让它在处理它之后返回一个值？

我也很高兴听到任何其他方法。

我试过这个样本https://dzone.com/articles/reactive-streams-in-java-9

我正在使用它进行自动交易。我对每只股票都有一个线程。每个线程都向策略发送消息...一旦策略成功，我希望它将结果发回给我

根据 The Definitive Guide to JasperReports – Teodor Danciu，Lucian Chirita，第 130 页，我在使用带有 Jasperreport 6.7.0 的 javaflow 时遇到子报表问题。由于有价值的原因，当报表在我们的应用程序中调用时，我需要它来避免不同的线程调用.

有设置在default.jasperreport.properties

当我运行超过 5 行的报告时出现错误：

java.lang.IllegalStateException：挂起时堆栈损坏（空堆栈）。net.sf.jasperreports.engine.fill.JRContinuationSubreportRunner@2024896527/org.apache.catalina.loader.WebappClassLoader@1847497893 是否用于 javaflow？在 org.apache.commons.javaflow.core.StackRecorder.execute(StackRecorder.java:131) 在 org.apache.commons.javaflow.api.Continuation$MultiShotContinuation.resumeWith(Continuation.java:521) 在 org.apache.commons .javaflow.api.Continuation.resume(Continuation.java:314) 在 org.apache.commons.javaflow.api.Continuation.startWith(Continuation.java:219) 在 org.apache.commons.javaflow.api.Continuation.startWith (Continuation.java:169) 在 org.apache.commons.javaflow.api.Continuation.startWith(Continuation.java:148) 在 net.sf.jasperreports.engine.fill.JRContinuationSubreportRunner。

如果行数较少，则报告运行没有问题。

我在 pom.xml 有依赖项来获取org.apache.commons.javaflow.api

将估计任何建议。

更新：我在 net.tascalate.javaflow.api 版本 2.2.1 中遇到了同样的错误

更新 2：我发现如果我们使 pageHeight 更大并且所有数据都适合一页，那么一切正常。所以问题与第二页初始化有关

在测试中，我想查看 HttpRequest 的主体内部。我想把身体作为一个字符串。似乎这样做的唯一方法是订阅 BodyPublisher ，但它是如何工作的？

是否可以让订阅者在与发布者相同的线程上运行（同步）？我可以使用 CompletableFuture，但它只提供一个结果。但是，如果我需要将许多结果交付给订阅者怎么办。请查看这个小测试以获得更好的解释。