Vert.x многопоточной веб-сокет

Question

У меня есть простое приложение vert.x:

public class Main { 
public static void main(String[] args) { 
    Vertx vertx = Vertx.vertx(new VertxOptions().setWorkerPoolSize(40).setInternalBlockingPoolSize(40)); 
    Router router = Router.router(vertx); 
    long main_pid = Thread.currentThread().getId(); 
    Handler<ServerWebSocket> wsHandler = serverWebSocket -> { 
     if(!serverWebSocket.path().equalsIgnoreCase("/ws")){ 
      serverWebSocket.reject(); 
     } else { 
      long socket_pid = Thread.currentThread().getId(); 
      serverWebSocket.handler(buffer -> { 
       String str = buffer.getString(0, buffer.length()); 
       long handler_pid = Thread.currentThread().getId(); 
       log.info("Got ws msg: " + str); 
       String res = String.format("(req:%s)main:%d sock:%d handlr:%d", str, main_pid, socket_pid, handler_pid); 
       try { 
        Thread.sleep(500); 
       } catch (InterruptedException e) { 
        e.printStackTrace(); 
       } 
       serverWebSocket.writeFinalTextFrame(res); 
      }); 
     } 
    }; 
    vertx 
     .createHttpServer() 
     .websocketHandler(wsHandler) 
     .listen(8080); 
} 
} 

При подключении этого сервера с несколькими клиентами, я вижу, что он работает в одном потоке. Но я хочу обрабатывать каждое клиентское соединение параллельно. Как я должен изменить этот код, чтобы сделать это?

Answer 1

Это:

новые VertxOptions() setWorkerPoolSize (40) .setInternalBlockingPoolSize (40)

Похоже, что вы пытаетесь создать свой собственный соединение HTTP пул, который, скорее всего, нет. что вы действительно хотите.

Идея Vert.x и другая неблокирующая event-loop рамки, основанной, в том, что мы не пытайтесь 1 thread -> 1 connection сродства, а, когда запрос, в настоящее время обслуживается нитью цикла обработки событий ждут IO - Е.Г. ответа из БД - поток потока событий освобождается для обслуживания другого соединения. Затем это позволяет потоку цикла одного потока обслуживать несколько соединений одновременно.

Если вы хотите полностью использовать все ядро ​​на своем компьютере, и вы будете запускать только один verticle, а затем установите количество экземпляров на количество ядер при развертывании вашей вертиляции.

IE

Vertx.vertx().deployVerticle("MyVerticle", new DeploymentOptions().setInstances(Runtime.getRuntime().availableProcessors())); 

Answer 2

Vert.x является reactive рамки, что означает, что она использует одну потоковую модель, чтобы удовлетворить все ваши загрузки приложений. Эта модель составляет known, чтобы масштабировать ее лучше, чем модель с резьбой.

Ключевым моментом является то, что весь код, который вы положили в обработчик , должен никогда не блокировать (как ваш Thread.sleep), так как он блокирует основной поток. Если блокирующий код (скажем, например, JDBC вызов), вы должны обернуть блокирующий код в executingBlocking обработчик, например:

serverWebSocket.handler(buffer -> { 
    String str = buffer.getString(0, buffer.length()); 
    long handler_pid = Thread.currentThread().getId(); 
    log.info("Got ws msg: " + str); 
    String res = String.format("(req:%s)main:%d sock:%d handlr:%d", str, main_pid, socket_pid, handler_pid); 
    vertx.executeBlocking(future -> { 
    try { 
     Thread.sleep(500); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } 
    serverWebSocket.writeFinalTextFrame(res); 
    future.complete(); 
    }); 
}); 

Теперь код блокировки будет работать в потоке из пула потоков, что вам можно настроить, как уже показано в других ответах.

Если вы хотите избежать написания всех этих обработчиков блокировки, и вы знаете, что вам нужно сделать несколько блокирующих вызовов, вам следует подумать об использовании worker verticle, так как они будут масштабироваться на уровне шины событий.

Заключительная записка для многопоточности заключается в том, что если вы используете несколько потоков, ваш сервер будет не таким эффективным, как один поток, например, он не сможет обрабатывать 10 миллионов веб-сайтов с 10 миллионов событий на современном машина (мы в 2016 году) поставит ваш планировщик ОС на колени.