Warn when SimpleAsyncTaskExecutor is used

Issue: SPR-16203
This commit is contained in:
Rossen Stoyanchev
2018-07-11 11:07:25 -04:00
parent 1b1bc7f5b5
commit 7b3a72f483
6 changed files with 96 additions and 37 deletions

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -24,10 +24,10 @@ import java.util.concurrent.Callable;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.web.context.request.async.CallableProcessingInterceptor;
import org.springframework.web.context.request.async.DeferredResult;
import org.springframework.web.context.request.async.DeferredResultProcessingInterceptor;
import org.springframework.web.context.request.async.WebAsyncTask;
/**
* Helps with configuring options for asynchronous request processing.
@@ -49,16 +49,15 @@ public class AsyncSupportConfigurer {
/**
* Set the default {@link AsyncTaskExecutor} to use when a controller method
* returns a {@link Callable}. Controller methods can override this default on
* a per-request basis by returning a {@link WebAsyncTask}.
* <p>By default a {@link SimpleAsyncTaskExecutor} instance is used, and it's
* highly recommended to change that default in production since the simple
* executor does not re-use threads.
* <p>As of 5.0 this executor is also used when a controller returns a reactive
* type that does streaming (e.g. "text/event-stream" or
* "application/stream+json") for the blocking writes to the
* {@link javax.servlet.ServletOutputStream}.
* The provided task executor is used to:
* <ol>
* <li>Handle {@link Callable} controller method return values.
* <li>Perform blocking writes when streaming to the response
* through a reactive (e.g. Reactor, RxJava) controller method return value.
* </ol>
* <p>By default only a {@link SimpleAsyncTaskExecutor} is used. However when
* using the above two use cases, it's recommended to configure an executor
* backed by a thread pool such as {@link ThreadPoolTaskExecutor}.
* @param taskExecutor the task executor instance to use by default
*/
public AsyncSupportConfigurer setTaskExecutor(AsyncTaskExecutor taskExecutor) {

View File

@@ -35,6 +35,7 @@ import org.springframework.core.MethodParameter;
import org.springframework.core.ReactiveAdapter;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.core.ResolvableType;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.http.MediaType;
@@ -79,6 +80,8 @@ class ReactiveTypeHandler {
private final TaskExecutor taskExecutor;
private Boolean taskExecutorWarning;
private final ContentNegotiationManager contentNegotiationManager;
@@ -92,6 +95,7 @@ class ReactiveTypeHandler {
Assert.notNull(manager, "ContentNegotiationManager is required");
this.reactiveRegistry = registry;
this.taskExecutor = executor;
this.taskExecutorWarning = executor instanceof SimpleAsyncTaskExecutor || executor instanceof SyncTaskExecutor;
this.contentNegotiationManager = manager;
}
@@ -127,16 +131,19 @@ class ReactiveTypeHandler {
if (adapter.isMultiValue()) {
if (mediaTypes.stream().anyMatch(MediaType.TEXT_EVENT_STREAM::includes) ||
ServerSentEvent.class.isAssignableFrom(elementClass)) {
logExecutorWarning(returnType);
SseEmitter emitter = new SseEmitter(STREAMING_TIMEOUT_VALUE);
new SseEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
return emitter;
}
if (CharSequence.class.isAssignableFrom(elementClass)) {
logExecutorWarning(returnType);
ResponseBodyEmitter emitter = getEmitter(mediaType.orElse(MediaType.TEXT_PLAIN));
new TextEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
return emitter;
}
if (mediaTypes.stream().anyMatch(MediaType.APPLICATION_STREAM_JSON::includes)) {
logExecutorWarning(returnType);
ResponseBodyEmitter emitter = getEmitter(MediaType.APPLICATION_STREAM_JSON);
new JsonEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
return emitter;
@@ -171,6 +178,27 @@ class ReactiveTypeHandler {
};
}
@SuppressWarnings("ConstantConditions")
private void logExecutorWarning(MethodParameter returnType) {
if (this.taskExecutorWarning && logger.isWarnEnabled()) {
synchronized (this) {
if (this.taskExecutorWarning) {
String executorTypeName = this.taskExecutor.getClass().getSimpleName();
logger.warn("\n!!!\n" +
"Streaming through a reactive type requires an Executor to write to the response.\n" +
"Please, configure a TaskExecutor in the MVC config under \"async support\".\n" +
"The " + executorTypeName + " currently in use is not suitable under load.\n" +
"-------------------------------\n" +
"Controller:\t" + returnType.getContainingClass().getName() + "\n" +
"Method:\t\t" + returnType.getMethod().getName() + "\n" +
"Returning:\t" + ResolvableType.forMethodParameter(returnType).toString() + "\n" +
"!!!");
this.taskExecutorWarning = false;
}
}
}
}
private abstract static class AbstractEmitterSubscriber implements Subscriber<Object>, Runnable {