从反应堆通量返回的错误变得无限答案

【问题标题】：Returning error from reactor Flux goes infinite从反应堆通量返回的错误变得无限
【发布时间】：2021-08-26 10:08:51
【问题描述】：

我已将 Micronaut 应用程序从 2.5.XX 更新到 3.0.0。根据 Micronaut 文档，项目 reactor 是推荐用于响应式编程的响应式库。

@Controller("/product")
public class ProductController implements IProductOperation {
    @Override
    public Flux<List> freeTextSearch(String text) {
        return iProductManager.findFreeText(text).onErrorResume(throwable -> {
            return Flux.error(new GlobalException(throwable));
        });
    }
}

public interface IProductOperation {
    @Get(value = "/search/{text}")
    Flux<?> freeTextSearch(@NotBlank String text);
}

当我 CURL 结束点 curl -X 'GET' 'http://localhost:8081/product/search/ffff' -H 'accept: application/json' 时，它会进入无限。

由于我在系统上有错误，所以 return Flux.error(new GlobalException(throwable)); 应该返回 500 内部系统错误，但是，它会进入无限期

我已经集成 rabbitMQ 并且 rabbitMQ 生产者已关闭。 iProductManager.findFreeText(text) 抛出异常，因为 rabbitMQ 生产者没有运行。它应该抛出一个异常并且应该调用全局错误处理，而不是无限。但它没有按预期工作

public class GlobalException extends RuntimeException{
    public GlobalException(Throwable throwable){super(throwable);}
}

这个方法永远不会在出错时被调用。

@Produces
@Singleton
@Requires(classes = {GlobalException.class, ExceptionHandler.class})
public class GlobalExceptionHandler implements ExceptionHandler<GlobalException, HttpResponse> {
    private static final Logger LOG = LoggerFactory.getLogger(GlobalExceptionHandler.class);

    @Override
    public HttpResponse handle(HttpRequest request, GlobalException exception) {
        LOG.error(exception.getLocalizedMessage());
        LOG.error(exception.getCause().getMessage());
        Arrays.stream(exception.getStackTrace()).forEach(item -> LOG.error(item.toString()));
        return HttpResponse.serverError(exception.getLocalizedMessage());
    }
}

日志

22:40:02.151 [default-nioEventLoopGroup-1-3] INFO  reactor.Flux.OnErrorResume.1 - onSubscribe(FluxOnErrorResume.ResumeSubscriber)
22:40:02.176 [default-nioEventLoopGroup-1-3] INFO  reactor.Flux.OnErrorResume.1 - request(1)

【问题讨论】：

标签： java project-reactor micronaut reactor micronaut-rest

【解决方案1】：

我认为您使用了错误的反应器Flux 运算符：

onErrorResume 切换到不同的Flux 以防出错；您可以将其视为“后备通量”。在您的情况下，回退会引发错误 - 这可以解释无限循环。
onErrorMap 应该满足您的要求：将异常映射到另一个可用于 HTTP 响应的异常

如果您想包装该 Flux 中的所有异常，您可以这样做：

return iProductManager.findFreeText(text)
        .onErrorMap(throwable -> new GlobalException(throwable));

请注意，其他 onErrorMap 方法允许您拥有更细粒度的行为。

【讨论】：

不，使用 onErrorMap 相同的问题不起作用
那么您应该使用调试日志更新您的问题，以便 Micronaut 找出递归发生的位置。您还可以在onErrorMap 之后添加.log() 运算符以获取更多信息。
添加了有关问题的日志，这些日志仅可用并且循环应用程序处于无限状态
为什么还在使用OnErrorResume？你切换到onErrorMap了吗？
我正在使用此代码返回 iProductManager.findFreeText(text).onErrorMap(throwable -> new GlobalException(throwable)).log();