multithreading au sein de vertx

Question

Je suis un débutant à vert.x. J'essayais la capacité "NetServer" de vert.x. http://vertx.io/core_manual_java.html#writing-tcp-servers-and-clients et cela fonctionne comme un charme.

Cependant, je lis aussi que « Une instance verticle est strictement un seul thread.

Si vous créez un serveur TCP simple et déployer une instance unique de celui-ci alors tous les gestionnaires de ce serveur sont toujours exécutées sur le même boucle d'événement (thread). "

Actuellement, pour ma mise en œuvre, je voulais recevoir le flux d'octets TCP et ensuite déclencher un autre composant. Mais cela ne devrait pas être un appel bloquant dans la méthode "start" de la Verticle. Donc, est-ce une bonne pratique, d'écrire un exécuteur dans la méthode de démarrage? ou est-ce que Vertx gère automatiquement ces cas?

Voici un extrait

public class TCPListener extends Verticle { 

    public void start(){ 

     NetServer server = vertx.createNetServer(); 

     server.connectHandler(new Handler<NetSocket>() { 
      public void handle(NetSocket sock) { 
       container.logger().info("A client has connected"); 
       sock.dataHandler(new Handler<Buffer>() { 
        public void handle(Buffer buffer) { 
         container.logger().info("I received " + buffer.length() + " bytes of data"); 

         container.logger().info("I received " + new String(buffer.getBytes())); 
         //Trigger another component here. SHould be done in a sperate thread. 
         //The previous call should be returned . No need to wait for component response. 
        } 
       }); 
      } 
     }).listen(1234, "host"); 
    } 
} 

Quel devrait être mécanisme pour faire un appel non bloquant.

Answer 1

Je ne pense pas que ce soit la voie à suivre pour vert.x.

Un meilleur moyen serait d'utiliser le bus d'événements correctement au lieu de Executor. Demandez à un travailleur de répondre à l'événement sur le bus, faites le traitement et signalez le bus lorsqu'il est terminé. La création de threads va à l'encontre du but d'aller avec vert.x.

Answer 2

La manière la plus flexible est de créer un ExecutorService et de traiter les demandes avec. Ceci apporte un contrôle fin sur le modèle de threads des travailleurs (nombre fixe ou variable de threads, quel travail doit être effectué en série sur un seul thread, etc.).

échantillon modifié pourrait ressembler à ceci:

public class TCPListener extends Verticle { 

    private final ExecutorService executor = Executors.newFixedThreadPool(10); 

    public void start(){ 

     NetServer server = vertx.createNetServer(); 

     server.connectHandler(new Handler<NetSocket>() { 
      public void handle(final NetSocket sock) { // <-- Note 'final' here 
       container.logger().info("A client has connected"); 
       sock.dataHandler(new Handler<Buffer>() { 
        public void handle(final Buffer buffer) { // <-- Note 'final' here 

         //Trigger another component here. SHould be done in a sperate thread. 
         //The previous call should be returned . No need to wait for component response. 
         executor.submit(new Runnable() { 

          public void run() { 
           //It's okay to read buffer data here 
           //and use sock.write() if necessary 
           container.logger().info("I received " + buffer.length() + " bytes of data"); 
           container.logger().info("I received " + new String(buffer.getBytes())); 
          } 
         } 
        } 
       }); 
      } 
     }).listen(1234, "host"); 
    } 
} 

Answer 3

Comme duffymo mentionné créer des discussions défaites dans le but d'utiliser vertx. Le meilleur moyen serait d'écrire un message dans eventbus et de créer un nouveau gestionnaire à l'écoute des messages du bus d'événements. Mise à jour du code pour montrer cela. Ecrire les messages dans la rubrique "next.topic" et enregistrer un gestionnaire pour lire le message de la rubrique "next.topic".

public class TCPListener extends Verticle { 

public void start(){ 

    NetServer server = vertx.createNetServer(); 

    server.connectHandler(new Handler<NetSocket>() { 
     public void handle(NetSocket sock) { 
      container.logger().info("A client has connected"); 
      sock.dataHandler(new Handler<Buffer>() { 
       public void handle(Buffer buffer) { 
        String recvMesg = new String(buffer.getBytes()); 
        container.logger().info("I received " + buffer.length() + " bytes of data"); 

        container.logger().info("I received " + recvMesg); 
        //Writing received message to event bus 
        vertx.eventBus().send("next.topic", recvMesg); 
       } 
      }); 
     } 
    }).listen(1234, "host"); 

    //Registering new handler listening to "next.topic" topic on event bus 
    vertx.eventBus().registerHandler("next.topic", new Handler<Message<String>() { 
     public void handle(Message<String> mesg) { 
      container.logger.info("Received message: "+mesg.body()); 
     } 
    }; 

} 
}