Reactor框架中轮询外部API的实现合理性与线程同步问题求助

问题 #

本人刚接触Reactor框架，若问题研究深度不足还请见谅。在我的响应式项目中，资源API接口返回Mono类型数据。某一阶段需要轮询外部API获取完成状态，再据此进行后续处理，最终构建资源并返回。此前我曾询问能否使用.repeatWhen()，但未达到预期效果，用户@amanin建议使用.retryWhen()，我最初对此有所顾虑，因此先尝试了.repeatWhen()。.retryWhen()能够按预期执行指定次数的重试并停止，以下是我适配后的getStatus()方法：

public Mono<AppContext> getStatus(AppContext ctx) {
    externalclient.getStatus() // external API call, client returns Mono<Response>
    .map(resp -> {
        if(resp.isCompleted()) {
            ctx.setCompleted(true);
            return resp;
        }
        throw new RetryException();
    })
    .retryWhen(Retry.backoff(maxAttempts).filter(err -> err instanceof RetryException))
    .subscribe();
    return Mono.just(ctx);
}

我有两个疑问：

1. 这种通过异常控制流程的方式是否属于反模式？是否为正确的实现方式？
2. 代码可正常执行，但功能测试中发现重试仍在进行时就返回了响应，主API线程与重试线程未关联，该如何修复？

解答 #

1. 异常控制流程是否为反模式？ #

在Reactor的重试机制设计中，通过特定异常触发重试是框架预期的用法，不算反模式，但要注意几个细节：

• 自定义的RetryException要明确语义，避免和业务异常、系统异常混淆，确保只有这个异常会触发重试逻辑。
• 可以把map里直接抛出异常的写法，改成用响应式API抛出错误，更贴合Reactor编程模型：

  .flatMap(resp -> {
      if(resp.isCompleted()) {
          ctx.setCompleted(true);
          return Mono.just(resp);
      }
      return Mono.error(new RetryException());
  })
  

• 禁止用通用异常（比如RuntimeException）触发重试，否则可能误重试非预期的错误场景。

2. 修复重试未完成就返回响应的问题 #

问题根源是你手动调用了.subscribe()，这会让外部API的调用流脱离当前响应式链，变成后台异步执行，而方法直接返回Mono.just(ctx)，自然会立刻响应。

正确做法是不要手动调用subscribe，把整个重试流和返回的Mono串联起来，让返回的Mono依赖重试流的完成信号：

public Mono<AppContext> getStatus(AppContext ctx) {
    return externalclient.getStatus()
        .map(resp -> {
            if(resp.isCompleted()) {
                ctx.setCompleted(true);
                return resp;
            }
            throw new RetryException();
        })
        .retryWhen(Retry.backoff(maxAttempts).filter(err -> err instanceof RetryException))
        .then(Mono.just(ctx)); // 重试完成后再返回ctx
}

这样整个流程会处于同一个响应式链中，主API的响应会等待重试流程完成后才发出，保证线程与流的关联。

内容的提问来源于stack exchange，提问作者xploreraj