Асинхронные сервлеты в Java: повышаем производительность приложений

<dependencies>
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>javax.servlet-api</artifactId>
<version>4.0.1</version>
<scope>provided</scope>
</dependency>

<!-- Опционально: для улучшенной работы с асинхронностью -->
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
<version>9.4.44.v20210927</version>
<scope>test</scope>
</dependency>

<!-- Для тестирования асинхронных компонентов -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13.2</version>
<scope>test</scope>
</dependency>
</dependencies>

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee 
http://xmlns.jcp.org/xml/ns/javaee/web-app_4_0.xsd"
version="4.0">

<servlet>
<servlet-name>AsyncServlet</servlet-name>
<servlet-class>com.example.AsyncServletExample</servlet-class>
<async-supported>true</async-supported>
</servlet>

<servlet-mapping>
<servlet-name>AsyncServlet</servlet-name>
<url-pattern>/async</url-pattern>
</servlet-mapping>

</web-app>

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.servlet.AsyncContext;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

@WebServlet(urlPatterns = "/async-example", asyncSupported = true)
public class BasicAsyncServlet extends HttpServlet {

// Создаем выделенный пул потоков для асинхронных задач
private final ExecutorService executorService = Executors.newFixedThreadPool(10);

@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) 
throws ServletException, IOException {

// Получаем асинхронный контекст
final AsyncContext asyncContext = request.startAsync();

// Опционально: устанавливаем таймаут (в миллисекундах)
asyncContext.setTimeout(30000);

// Передаем асинхронную задачу в пул потоков
executorService.submit(() -> processAsyncRequest(asyncContext));

// Метод doGet возвращается немедленно, освобождая поток контейнера
System.out.println("Servlet thread released");
}

private void processAsyncRequest(AsyncContext asyncContext) {
try {
// Имитация длительной операции
System.out.println("Long-running task started");
Thread.sleep(5000);
System.out.println("Long-running task completed");

// Получаем response из контекста
HttpServletResponse response = 
(HttpServletResponse) asyncContext.getResponse();

// Устанавливаем заголовки и тип контента
response.setContentType("text/plain");
response.setCharacterEncoding("UTF-8");

// Записываем ответ
response.getWriter().write("Async processing completed at: " + 
new java.util.Date());

} catch (Exception e) {
System.err.println("Error in async processing: " + e.getMessage());
} finally {
// Важно: завершаем асинхронную обработку
asyncContext.complete();
}
}

@Override
public void destroy() {
executorService.shutdown();
super.destroy();
}
}

import javax.servlet.AsyncEvent;
import javax.servlet.AsyncListener;
import java.io.IOException;

public class CustomAsyncListener implements AsyncListener {

@Override
public void onComplete(AsyncEvent event) throws IOException {
System.out.println("Async request completed");
}

@Override
public void onTimeout(AsyncEvent event) throws IOException {
System.out.println("Async request timed out");
// Обработка случая превышения времени ожидания
event.getAsyncContext().getResponse().getWriter()
.write("Operation timed out after " + 
event.getAsyncContext().getTimeout() + " ms");
event.getAsyncContext().complete();
}

@Override
public void onError(AsyncEvent event) throws IOException {
System.out.println("Error in async request: " + event.getThrowable().getMessage());
// Обработка ошибки
event.getThrowable().printStackTrace();
event.getAsyncContext().complete();
}

@Override
public void onStartAsync(AsyncEvent event) throws IOException {
System.out.println("Async request started");
}
}

@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) 
throws ServletException, IOException {

// Получаем асинхронный контекст
final AsyncContext asyncContext = request.startAsync();

// Устанавливаем таймаут
asyncContext.setTimeout(30000);

// Добавляем слушателя событий
asyncContext.addListener(new CustomAsyncListener());

// Передаем асинхронную задачу в пул потоков
executorService.submit(() -> processAsyncRequest(asyncContext));

System.out.println("Servlet thread released");
}

private void processAsyncRequest(AsyncContext asyncContext) {
HttpServletResponse response = null;
try {
// Имитация длительной операции
System.out.println("Long-running task started");
Thread.sleep(5000);
System.out.println("Long-running task completed");

// Получаем response из контекста
response = (HttpServletResponse) asyncContext.getResponse();

// Проверка, не был ли запрос уже завершен (например, по таймауту)
if (asyncContext.getRequest().isAsyncStarted()) {
// Устанавливаем заголовки и тип контента
response.setContentType("text/plain");
response.setCharacterEncoding("UTF-8");

// Записываем ответ
response.getWriter().write("Async processing completed at: " + 
new java.util.Date());
}

} catch (Exception e) {
System.err.println("Error in async processing: " + e.getMessage());
// Обработка ошибки, если ответ еще не отправлен
if (response != null && asyncContext.getRequest().isAsyncStarted()) {
try {
response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
response.getWriter().write("Error occurred: " + e.getMessage());
} catch (IOException ioe) {
System.err.println("Failed to write error response: " + ioe.getMessage());
}
}
} finally {
// Завершаем асинхронную обработку, если она еще активна
if (asyncContext.getRequest().isAsyncStarted()) {
asyncContext.complete();
}
}
}

@WebServlet(urlPatterns = "/async-executor", asyncSupported = true)
public class ExecutorAsyncServlet extends HttpServlet {

private final ExecutorService executorService = Executors.newFixedThreadPool(20);

@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) 
throws ServletException, IOException {

final AsyncContext asyncContext = request.startAsync();
asyncContext.setTimeout(60000); // 60 секунд

executorService.execute(() -> {
try {
// Имитация сложного запроса к базе данных
Thread.sleep(3000);

// Получение данных
String result = fetchDataFromDatabase();

HttpServletResponse asyncResponse = 
(HttpServletResponse) asyncContext.getResponse();
asyncResponse.setContentType("application/json");
asyncResponse.getWriter().write(result);

} catch (Exception e) {
log("Error in async processing", e);
} finally {
asyncContext.complete();
}
});
}

private String fetchDataFromDatabase() {
// Здесь был бы реальный запрос к БД
return "{\"status\":\"success\",\"data\":[1,2,3]}";
}

@Override
public void destroy() {
executorService.shutdown();
try {
if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
executorService.shutdownNow();
}
} catch (InterruptedException e) {
executorService.shutdownNow();
}
super.destroy();
}
}

@WebServlet(urlPatterns = "/async-completable", asyncSupported = true)
public class CompletableFutureServlet extends HttpServlet {

private final ExecutorService executorService = Executors.newFixedThreadPool(10);
private final ExternalServiceClient serviceClient = new ExternalServiceClient();

@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) 
throws ServletException, IOException {

final String userId = request.getParameter("userId");
final AsyncContext asyncContext = request.startAsync();

// Создаем цепочку асинхронных операций
CompletableFuture.supplyAsync(() -> serviceClient.getUserProfile(userId), executorService)
.thenCombine(
CompletableFuture.supplyAsync(() -> serviceClient.getUserOrders(userId), executorService),
(profile, orders) -> {
// Объединяем результаты двух параллельных запросов
Map<String, Object> result = new HashMap<>();
result.put("profile", profile);
result.put("orders", orders);
return result;
}
)
.thenAccept(result -> {
try {
HttpServletResponse asyncResponse = 
(HttpServletResponse) asyncContext.getResponse();
asyncResponse.setContentType("application/json");
// В реальном приложении использовали бы Jackson или Gson
asyncResponse.getWriter().write(result.toString());
} catch (IOException e) {
throw new CompletionException(e);
} finally {
asyncContext.complete();
}
})
.exceptionally(ex -> {
try {
HttpServletResponse asyncResponse = 
(HttpServletResponse) asyncContext.getResponse();
asyncResponse.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
asyncResponse.getWriter().write("Error: " + ex.getMessage());
} catch (IOException e) {
log("Failed to send error response", e);
} finally {
asyncContext.complete();
}
return null;
});
}

// Имитация клиента внешнего сервиса
static class ExternalServiceClient {
public Map<String, Object> getUserProfile(String userId) {
// Имитация внешнего вызова
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}

Map<String, Object> profile = new HashMap<>();
profile.put("id", userId);
profile.put("name", "User " + userId);
return profile;
}

public List<String> getUserOrders(String userId) {
// Имитация внешнего вызова
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}

return Arrays.asList("Order1", "Order2", "Order3");
}
}

@Override
public void destroy() {
executorService.shutdown();
super.destroy();
}
}

@WebServlet(urlPatterns = "/stream-data", asyncSupported = true)
public class StreamingServlet extends HttpServlet {

private final ScheduledExecutorService scheduler = 
Executors.newScheduledThreadPool(1);

@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) 
throws ServletException, IOException {

final AsyncContext asyncContext = request.startAsync();

// Настройка для SSE
HttpServletResponse asyncResponse = (HttpServletResponse) asyncContext.getResponse();
asyncResponse.setContentType("text/event-stream");
asyncResponse.setCharacterEncoding("UTF-8");
asyncResponse.setHeader("Cache-Control", "no-cache");

// Счетчик для демонстрации
final AtomicInteger counter = new AtomicInteger(0);

// Планируем периодическую отправку данных
final ScheduledFuture<?> task = scheduler.scheduleAtFixedRate(() -> {
try {
if (counter.get() >= 10) {
asyncContext.complete();
return;
}

PrintWriter writer = asyncResponse.getWriter();
writer.write("data: " + new Date() + " – Update " + 
counter.incrementAndGet() + "\n\n");
writer.flush();

} catch (IOException e) {
asyncContext.complete();
}
}, 0, 1, TimeUnit.SECONDS);

// Добавляем слушателя для отмены задачи при прерывании соединения
asyncContext.addListener(new AsyncListener() {
@Override
public void onComplete(AsyncEvent event) {
task.cancel(false);
}

@Override
public void onTimeout(AsyncEvent event) {
task.cancel(false);
}

@Override
public void onError(AsyncEvent event) {
task.cancel(false);
}

@Override
public void onStartAsync(AsyncEvent event) {}
});
}

@Override
public void destroy() {
scheduler.shutdown();
super.destroy();
}
}

// Для CPU-интенсивных операций (число потоков равно количеству ядер)
int cpuCores = Runtime.getRuntime().availableProcessors();
ExecutorService computeExecutor = new ThreadPoolExecutor(
cpuCores, cpuCores,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(1000),
new ThreadFactory() {
final AtomicInteger counter = new AtomicInteger();
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("compute-worker-" + counter.incrementAndGet());
thread.setPriority(Thread.NORM_PRIORITY);
return thread;
}
},
new ThreadPoolExecutor.CallerRunsPolicy()
);

// Для I/O-интенсивных операций (число потоков больше, т.к. операции в основном ждут)
ExecutorService ioExecutor = new ThreadPoolExecutor(
cpuCores * 2, cpuCores * 10,
60L, TimeUnit.SECONDS,
new SynchronousQueue<>(),
new ThreadFactory() {
final AtomicInteger counter = new AtomicInteger();
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("io-worker-" + counter.incrementAndGet());
thread.setDaemon(true);
return thread;
}
},
new ThreadPoolExecutor.AbortPolicy()
);

@WebServlet(urlPatterns = "/async-pipeline", asyncSupported = true)
public class PipelineServlet extends HttpServlet {

@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) 
throws ServletException, IOException {

// Начинаем асинхронную обработку
final AsyncContext asyncContext = request.startAsync();

// Запускаем первую стадию обработки
stage1(asyncContext);
}

private void stage1(AsyncContext context) {
ioExecutor.execute(() -> {
try {
// Получение данных из БД
String data = fetchFromDatabase();

// Сохраняем результат в атрибутах для следующего этапа
context.getRequest().setAttribute("data", data);

// Переходим к следующему этапу в другом пуле потоков
stage2(context);

} catch (Exception e) {
handleError(context, e);
}
});
}

private void stage2(AsyncContext context) {
computeExecutor.execute(() -> {
try {
// Получаем данные из предыдущего этапа
String data = (String) context.getRequest().getAttribute("data");

// Обрабатываем данные (CPU-интенсивная операция)
String processedData = processData(data);

// Переходим к финальному этапу
complete(context, processedData);

} catch (Exception e) {
handleError(context, e);
}
});
}

private void complete(AsyncContext context, String result) {
try {
HttpServletResponse response = (HttpServletResponse) context.getResponse();
response.setContentType("application/json");
response.getWriter().write("{\"result\":\"" + result + "\"}");
} catch (IOException e) {
log("Error writing response", e);
} finally {
context.complete();
}
}

private void handleError(AsyncContext context, Exception e) {
try {
log("Error in async processing", e);
HttpServletResponse response = (HttpServletResponse) context.getResponse();
response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
response.getWriter().write("Error: " + e.getMessage());
} catch (IOException ioe) {
log("Failed to write error response", ioe);
} finally {
context.complete();
}
}

// Методы имитации операций опущены для краткости
}

@WebServlet(urlPatterns = "/async-dispatch", asyncSupported = true)
public class DispatchingServlet extends HttpServlet {

@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) 
throws ServletException, IOException {

final AsyncContext asyncContext = request.startAsync();

executorService.execute(() -> {
try {
// Выполняем асинхронную операцию
Map<String, Object> result = performComplexOperation();

// Сохраняем результат в атрибутах запроса
request.setAttribute("data", result);

// Перенаправляем запрос на JSP для отображения результатов
asyncContext.dispatch("/WEB-INF/views/result.jsp");

} catch (Exception e) {
log("Error in async processing", e);
asyncContext.complete();
}
});
}
}

@WebServlet(urlPatterns = "/resource-management", asyncSupported = true)
public class ResourceManagementServlet extends HttpServlet {

@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) 
throws ServletException, IOException {

final AsyncContext asyncContext = request.startAsync();

// Устанавливаем разумный таймаут
asyncContext.setTimeout(30000);

// Создаем ресурсы, требующие очистки
final Connection dbConnection = getDatabaseConnection();

// Добавляем обработчик событий для гарантированной очистки ресурсов
asyncContext.addListener(new AsyncListener() {
@Override
public void onComplete(AsyncEvent event) {
closeResources(dbConnection);
}

@Override
public void onTimeout(AsyncEvent event) {
log("Request timed out");
handleTimeout((HttpServletResponse) event.getSuppliedResponse());
closeResources(dbConnection);
}

@Override
public void onError(AsyncEvent event) {
log("Error occurred", event.getThrowable());
closeResources(dbConnection);
}

@Override
public void onStartAsync(AsyncEvent event) {
// Обычно не требует специальной обработки
}

private void closeResources(Connection conn) {
try {
if (conn != null && !conn.isClosed()) {
conn.close();
}
} catch (SQLException e) {
log("Error closing database connection", e);
}
}

private void handleTimeout(HttpServletResponse response) {
try {
response.setStatus(HttpServletResponse.SC_GATEWAY_TIMEOUT);
response.getWriter().write("Request processing timed out");
} catch (IOException e) {
log("Failed to write timeout response", e);
}
}
});

// Продолжаем асинхронную обработку
executorService.execute(() -> processRequestWithConnection(asyncContext, dbConnection));
}

// Другие методы опущены для краткости
}

@WebServlet(urlPatterns = "/monitored-async", asyncSupported = true)
public class MonitoredAsyncServlet extends HttpServlet {

private final MetricsRegistry metricsRegistry = new MetricsRegistry();

@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) 
throws ServletException, IOException {

// Начинаем отсчёт времени
long startTime = System.currentTimeMillis();
String operationType = request.getParameter("operation");

final AsyncContext asyncContext = request.startAsync();

// Увеличиваем счётчик активных запросов
metricsRegistry.incrementActiveRequests();

executorService.execute(() -> {
try {
// Выполняем операцию
String result = performOperation(operationType);

// Записываем ответ
HttpServletResponse asyncResponse = 
(HttpServletResponse) asyncContext.getResponse();
asyncResponse.setContentType("text/plain");
asyncResponse.getWriter().write(result);

} catch (Exception e) {
log("Error in async processing", e);
metricsRegistry.incrementErrorCount(operationType);
} finally {
// Обновляем метрики
long processingTime = System.currentTimeMillis() – startTime;
metricsRegistry.recordOperationTime(operationType, processingTime);
metricsRegistry.decrementActiveRequests();

asyn