如何在 Reactor 中使用 Mono 的内容

Posted 2023-02-16

【中文标题】如何在 Reactor 中使用 Mono 的内容

【英文标题】：How to use Mono's content in Reactor

【发布时间】：2021-12-15 15:28:47

【问题描述】：

我正在尝试调用反应式 REST API 来获取截止日期TS 变量。 然后我试图在我的 Pojo 类中设置相同的值。 但是在 BOLCompliance 中设置了 deadlineTS 后的值与使用 subscribe() 不一致。有时我可以设置值，有时我得到空值。如何确保每次都能设置值。

Mono<String> deadlineTS = portCallServiceCaller.getDeadlineTSByComplianceId(compliance.getId());

BOLCompliance complianceResponse = new BOLCompliance();
deadlineTS.subscribe(val->complianceResponse.setDeadlineTimestamp(val));

【问题讨论】：

@lkatiforis: 请回答

【参考方案1】：

在反应式管道之外使用 Mono 的内容（阻塞）

您可以像这样使用block() 方法：

Mono<String> nameMono = Mono.just("some-value").delayElement(Duration.ofMillis(300));

Person person = new Person();
person.setName(nameMono.block());

System.out.println(person.getName());

这会触发操作并等待其完成。注意调用线程阻塞。

或者，您可以使用 subscribe(consumer, errorConsumer, completeConsumer) 并提供一个将在操作完成时触发的 Runnable：

valueMono.subscribe(v-> person.setName(v), throwable -> , () -> System.out.println(person.getName()));

但是，subscribe() 方法会立即返回。

在反应式管道中使用 Mono 的内容

您可以根据情况选择提供的运算符之一。

在这种情况下，您可以使用map 运算符将String 转换为BOLCompliance：

  Mono<BOLCompliance> fetchBOLCompliance() 
    Mono<String> deadlineMono = portCallServiceCaller.getDeadlineTSByComplianceId(compliance.getId();

    return deadlineMono.map(deadline -> 
      BOLCompliance compliance = new BOLCompliance();
      compliance.setDeadlineTimestamp(deadline);
      return compliance;
    );
  

如果您想运行异步任务（例如数据库访问），您需要使用flatmap 运算符。

subscribe() 立即返回

根据Javadoc：

Disposable subscribe(Consumer<? super T> consumer)

请记住，由于序列可以是异步的，这将 立即将控制权返回给调用线程。这可以给 在主线程中执行时未调用消费者的印象 或例如单元测试。

换句话说，subscribe 方法会启动奇怪的工作并立即返回。因此，您无法保证操作已完成。例如，以下示例将始终以空值结束：

Mono<String> nameMono = Mono.just("some-value").delayElement(Duration.ofMillis(300));

Person person = new Person();
nameMono.subscribe(v-> person.setName(v));

System.out.println(person.getName());

这里，person.getName() 方法被立即调用，而person.setName(v) 在 300 毫秒后被调用。

【讨论】：

如果我使用 block()，我会收到这个错误 - block()/blockFirst()/blockLast() 正在阻塞，线程反应器不支持，现在如何修复？

已编辑。你不应该在反应式管道中使用block()。

@MiraDevi 这对你有用吗？如果它回答了您的问题，那么请考虑投票并接受它为正确的答案，以便其他人可以从中受益并轻松理解这可能是类似问题的可能解决方案。

是的，确实如此，很抱歉没有投票，因为我在想问这个问题的人，他的投票不算数。非常感谢您的解决方案。