如何为Webflux应用程序包装阻塞IO操作

问题描述

我有一个Spring Boot / Webflux应用程序。我需要一个执行以下操作的异步REST端点：

1. 生成随机作业ID。
2. 通过HTTP调用一项服务。
3. 通过HTTP调用其他服务。
4. 合并来自服务的响应并将结果写入文件。
5. 将工作ID返回给客户端。

端点应该是异步的。这就是为什么客户不应该等待步骤2、3、4的结果。客户应立即收到工作ID。

现在我有以下实现：

@Override
  public Mono<String> saveData() {
    String jobId = UUID.randomUUID().toString();
    Mono<ResponseFromService1> response1 = service1.getData();
    Mono<ResponseFromService2> response2 = service2.getData();
    return fileService.saveData(response1,response2)
        .map(filePath -> log.info("File has been stored at {}",filePath))
        .map(jobId);

service1和service2通过使用反应式WebClient来实现。 fileService.saveData的实现如下：

  public Mono<Path> saveDataInFile(Mono<ResponseFromService1> response1,Mono<ResponseFromService2> response2) {
return Mono.fromCallable(() ->
    Mono.zip(response1,response2)
        .map(tuple -> blockingIOsaveMethod(tuple.getT1(),tuple.getT2()))
).publishOn(Schedulers.elastic())
    .flatMap(mono -> mono);
  }

问题在于此端点不是异步的。保存包含数据的文件后，端点的客户端将获得作业ID。 我应该如何更新saveDataInFile和saveData以立即返回作业ID？

解决方法

客户应立即收到工作ID。

这似乎表明Mono<String>不是saveData()的正确返回类型，因为您不想等待任何异步操作明显完成：

@Override
public String saveData() {
  String jobId = UUID.randomUUID().toString();
  // ...
  return jobId;
}

...看起来像“发射后遗忘”操作，您可以手动订阅Mono：

@Override
public String saveData() {
  String jobId = UUID.randomUUID().toString();
  fileService.saveData(...).subscribe(...); // look at the different overloads of #subscribe(...)
  return jobId;
}

请确保所有内容均已正确记录，以免跟踪HTTP响应后发生的情况。