spring mvc 与反应流集成答案

【问题标题】：spring mvc integrate with reactive streamspring mvc 与反应流集成
【发布时间】：2016-05-12 16:26:10
【问题描述】：

我有一个基于 spring mvc 构建的 RESTful api 应用程序。

最近我在 spring mvc 和响应式流（如 rxjava 和 project-reactor）之间进行了一些集成，并尝试使应用程序更具响应性。

我刚刚构建了一些如下所示的演示：

1.对于 rxjava，我使用 PublishSubject

private SerializedSubject<StreamResult, StreamResult> subject = PublishSubject.<StreamResult>create().toSerialized();

public ReactiveStreamController() {
    this.subject.subscribe(streamResult -> {
        String id = streamResult.getRequest().getParameter("id");
        System.out.println("[" + Thread.currentThread().getName() + "] request received. id = " + id);
        String random = StringUtils.isBlank(id) ? StringUtils.EMPTY : id;
        ResponseVO vo = new ResponseVO(200, "success = " + random);
        streamResult.getFuture().complete(vo);
    }, Throwable::printStackTrace);
}

@ResponseBody
@RequestMapping(value = "/rxJava", method = RequestMethod.GET)
public CompletableFuture<ResponseVO> rxJavaController(HttpServletRequest httpServletRequest) {
    StreamResult sr = new StreamResult();
    sr.setRequest(httpServletRequest);
    subject.onNext(sr);
    return sr.getFuture();
}

2.用于项目反应堆

@ResponseBody
@RequestMapping(value = "/reactorCodeNew", method = RequestMethod.GET)
public CompletableFuture<ResponseVO> reactorCoreNewParadigm(HttpServletRequest servletRequest) {
    Mono<ResponseVO> mono = Mono.just(servletRequest)
            .subscribeOn(executorService)
            .map(request -> {
                String id = request.getParameter("id");
                System.out.println("[" + Thread.currentThread().getName() + "] request received. id = " + id);
                String random = StringUtils.isBlank(id) ? StringUtils.EMPTY : id;
                ResponseVO vo = new ResponseVO(200, "success = " + random);
                return vo;
            })
            .timeout(Duration.ofSeconds(2), Mono.just(new ResponseVO(500, "error")));
    return mono.toCompletableFuture();
}

在运行这两个演示时，我看不出仅使用 java 的 CompletableFuture 来提供控制器方法之间有太多区别。

我所理解的响应式流和我想要的是将 servlet 请求视为流并将其与背压等功能结合起来。

我想知道： 1.有没有更好的方法让应用程序更具反应性？ 2.将spring mvc与反应流集成是否正确或兼容？如果是，我怎样才能执行诸如背压之类的功能？

我意识到也许我忘了声明为什么/如何在控制器中返回一个可完成的未来，实际上我注入了一个自定义的 MethodReturnValueHandler 来将 CompletableFuture 转换为 DefferdResult。

public class CompletableFutureMethodReturnValueHandler extends DeferredResultMethodReturnValueHandler {

@Override
public boolean supportsReturnType(MethodParameter returnType) {
    return CompletableFuture.class.isAssignableFrom(returnType.getParameterType());
}

@Override
public void handleReturnValue(Object returnValue, MethodParameter returnType, ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {

    CompletableFuture<?> completableFuture = (CompletableFuture<?>) returnValue;

    super.handleReturnValue(CompletableDeferredResult.newInstance(completableFuture), returnType, mavContainer, webRequest);
}
}

【问题讨论】：

最近在此发布的 Spring 博客：spring.io/blog/2016/07/20/…

标签： java spring-mvc rx-java reactive-programming project-reactor

【解决方案1】：

Spring MVC 基于 Servlet API 并且大部分是在内部阻塞的，因此它不能利用响应式流的行为。为 Controller 层编写适配器是不够的。

为此，Spring 团队正在制定一项单独的计划。关注 SPR-14161 和 Spring 博客（包括 this 和 this）以了解有关响应式 Spring 的更多信息。

【讨论】：