如何为 undertow 服务器侦听器设置多个工作线程？

问题描述

我正在尝试为 undertow 服务器创建多个工作线程，以便多个线程可以同时处理来自客户端的请求。 我尝试使用 UndertowOptions 和 Options 在服务器级别设置它们。尝试创建自定义 XnioWorker 并分配给 Undertow Server。

这里似乎没有任何工作，所有尝试配置工作线程。

Xnio xnio = Xnio.getInstance();
        XnioWorker worker = null;
        try {
            worker = xnio.createWorker(
                    OptionMap.builder().set(Options.WORKER_IO_THREADS,50).set(Options.WORKER_TASK_CORE_THREADS,50)
                            .set(Options.WORKER_TASK_MAX_THREADS,50).set(Options.TCP_NODELAY,true).getMap());
        } catch (Exception e) {
            // Todo Auto-generated catch block
            e.printstacktrace();
        }
//      OptionMap socketoptions = OptionMap.builder().set(Options.WORKER_IO_THREADS,10).set(Options.TCP_NODELAY,true)
//              .set(Options.REUSE_ADDRESSES,true).getMap();

        Undertow server = Undertow.builder().setServerOption(UndertowOptions.ENABLE_HTTP2,true)
                .setServerOption(UndertowOptions.HTTP2_SETTINGS_MAX_CONCURRENT_STREAMS,UndertowHttp2ServerEpLevel.maxReqPerConn)
                .setServerOption(UndertowOptions.MAX_CONCURRENT_REQUESTS_PER_CONNECTION,UndertowHttp2ServerEpLevel.maxReqPerConn)
                .addHttpListener(port,host)
                .setWorkerOption(Options.WORKER_IO_THREADS,10)
                .setWorkerOption(Options.WORKER_TASK_CORE_THREADS,10)
                .setWorkerThreads(UndertowHttp2ServerEpLevel.iotheads)
                .setIoThreads(UndertowHttp2ServerEpLevel.iotheads)
                .setWorker(worker)
                .setHandler(exchange -> {

                    Thread.sleep(responseDelayInMs);

                    DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
                    Date date = new Date();

                    System.out.print(dateFormat.format(date) + ",**Thread name: " + Thread.currentThread().getName());**
                    System.out.println(
                            ": " + port + ",Client address is: " + exchange.getConnection().getPeerAddress().toString()
                                    + " TotalMsgReceived: " + ++msgCounter);

                    exchange.getResponseHeaders().put(Headers.CONTENT_TYPE,"application/json");
                    exchange.setStatusCode(returnCode);
//                  exchange.getResponseSender().send("Undertow Hi");
               
                }).build();
        server.start();

通过上面的代码，我看到处理程序代码总是由同一个线程执行。并且假设一个线程的 n/w 延迟为 100 毫秒，我每秒只能处理 9 个请求。这就是为什么我想要更多的线程来并发处理请求，以实现高速率。

控制台输出：

*2021/07/29 09:22:46，线程名称：XNIO-1 I/O-12：8081，客户端地址：/127.0.0.1:16002 TotalMsgReceived：534

2021/07/29 09:22:47，线程名称：XNIO-1 I/O-12：8081，客户端地址：/127.0.0.1:16002 TotalMsgReceived：535*

现在从 JMX 中，我看到线程池大小已更新为 50，但性能仍然没有改善，控制台中也没有看到多工作线程：

对此的任何看法都会有所帮助。谢谢！

解决方法

来自documentation

分派到工作线程

public void handleRequest(final HttpServerExchange exchange) throws Exception{
    if (exchange.isInIoThread()) {
      exchange.dispatch(this);
      return;
    }
    //handler code
}

您似乎遗漏了这一部分，因为您的案例中的所有内容都在 IO 线程上运行。

您可以根据自己的用例选择 multiple overloads of dispatch method。

,

以下代码可以完成工作。

Undertow server = Undertow.builder().setServerOption(UndertowOptions.ENABLE_HTTP2,true)
                .setServerOption(UndertowOptions.HTTP2_SETTINGS_MAX_CONCURRENT_STREAMS,UndertowHttp2ServerEpLevelMT.maxReqPerConn)
                .setServerOption(UndertowOptions.MAX_CONCURRENT_REQUESTS_PER_CONNECTION,UndertowHttp2ServerEpLevelMT.maxReqPerConn)
                .addHttpListener(port,host)
                .setWorkerOption(Options.WORKER_IO_THREADS,UndertowHttp2ServerEpLevelMT.iotheads)
                .setWorkerOption(Options.WORKER_TASK_CORE_THREADS,UndertowHttp2ServerEpLevelMT.iotheads)
                .setWorkerOption(Options.WORKER_TASK_MAX_THREADS,UndertowHttp2ServerEpLevelMT.iotheads)
                .setWorkerThreads(UndertowHttp2ServerEpLevelMT.iotheads)
                .setIoThreads(UndertowHttp2ServerEpLevelMT.iotheads)
                .setHandler(new HttpHandler() {
                    @Override
                    public void handleRequest(final HttpServerExchange exchange) throws Exception {
                        if (exchange.isInIoThread()) {
                            exchange.dispatch(this);
                            return;
                        }
                        System.out.println("Port: " + port + ",Client address is: "
                                + exchange.getConnection().getPeerAddress().toString());
                        exchange.setStatusCode(returnCode);
                    }
                }).build();
        server.start();