暗流。与另一个线程的异步响应

Posted 2023-02-27

【中文标题】暗流。与另一个线程的异步响应

【英文标题】：Undertown. Async responce with another thread

【发布时间】：2018-05-07 23:17:11

【问题描述】：

我尝试了解如何使用 Undertow 构建一个真正的异步 http 服务器。如果我有另一个线程已经在处理请求，如何异步发送响应？ 我写了这样的代码：

    Undertow server = Undertow.builder()
            .addHttpListener(8080, "localhost")
            .setHandler(exchange -> 
                CompletableFuture.runAsync(() -> 
                    try 
                        Thread.sleep(100);
                     catch (InterruptedException e) 
                        throw new RuntimeException(e);
                    
                ).thenAccept(string -> 
                    exchange.getResponseHeaders()
                            .put(Headers.CONTENT_TYPE, "text/plain");
                    exchange.getResponseSender().send("Hello World");
                    exchange.endExchange();
                ).exceptionally(throwable -> 
                    System.out.println(throwable.toString());
                    return null;
                );
            ).build();
    server.start();

但是这个服务器响应 200 没有数据并且在日志中

java.lang.IllegalStateException: UT000127: 响应已经发送

当我像这样使用 io.undertow.server.HttpServerExchange#dispatch(java.lang.Runnable) 方法时：

    Undertow server = Undertow.builder()
            .addHttpListener(8080, "localhost")
            .setHandler(exchange -> 

                exchange.dispatch(() -> 

                    CompletableFuture.runAsync(() -> 
                        try 
                            Thread.sleep(100);
                         catch (InterruptedException e) 
                            throw new RuntimeException(e);
                        
                    ).thenAccept(string -> 
                        exchange.getResponseHeaders()
                                .put(Headers.CONTENT_TYPE,"text/plain");
                        exchange.getResponseSender().send("Hello World");
                        exchange.endExchange();
                    ).exceptionally(throwable -> 
                        System.out.println(throwable.toString());
                        return null;
                    );

                );
            ).build();
    server.start();

当然响应“Hello World”，但服务器会为每个请求创建新线程！

(10 个并行请求后的 jvisualvm)

【问题讨论】：

P.S.我在 main 方法中运行它。

【参考方案1】：

暗流不支持这种方式，

我创建了一个新项目来解决它：

https://github.com/hank-whu/undertow-async

package io.undertow.async.pingpong;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;

import io.undertow.async.handler.AsyncHttpHandler;
import io.undertow.async.io.PooledByteBufferInputStream;
import io.undertow.async.io.PooledByteBufferOutputStream;
import io.undertow.connector.ByteBufferPool;
import io.undertow.server.HttpServerExchange;
import io.undertow.util.StatusCodes;

public class PingPongAsyncHttpHandler extends AsyncHttpHandler 

    @Override
    protected void handleAsyncRequest(HttpServerExchange exchange, PooledByteBufferInputStream content)
            throws Exception 

        CompletableFuture//
                .completedFuture(content)// init
                .thenApplyAsync(this::readBytesAndClose)// read
                .thenApplyAsync(bytes -> // write
                    ByteBufferPool byteBufferPool = exchange.getConnection().getByteBufferPool();
                    PooledByteBufferOutputStream output = new PooledByteBufferOutputStream(byteBufferPool);
                    write(output, bytes);
                    return output;
                )//
                .thenAcceptAsync(output -> send(exchange, StatusCodes.OK, output));
    

    private byte[] readBytesAndClose(PooledByteBufferInputStream content) 
        try 
            byte[] bytes = new byte[content.available()];
            content.read(bytes);
            return bytes;
         catch (IOException e) 
            throw new RuntimeException(e);
         finally 
            try // must close it
                content.close();
             catch (IOException e) 
                e.printStackTrace();
            
        
    

    private void write(PooledByteBufferOutputStream output, byte[] bytes) 
        try 
            output.write("asycn response: ");
            output.write(bytes);
         catch (IOException e) 
            throw new RuntimeException(e);