我有一个基于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. 对于Reactor项目
@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与响应式流集成？如果可以，如何执行背压等特性？
我意识到可能忘记声明为什么/如何在控制器中返回CompletableFuture，实际上我注入了一个自定义的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);
}
}