2

私は vert.x の初心者です。私は vert.x の「NetServer」機能を試していました。http://vertx.io/core_manual_java.html#writing-tcp-servers-and-clientsとそれは魅力のように動作します。

ただし、「バーティクル インスタンスは厳密にシングル スレッドです。

単純な TCP サーバーを作成し、そのインスタンスを 1 つデプロイすると、そのサーバーのすべてのハンドラーが常に同じイベント ループ (スレッド) で実行されます。」

現在、私の実装では、バイトの TCP ストリームを受信して​​から、別のコンポーネントをトリガーしたいと考えていました。ただし、これは Verticle の「開始」メソッド内でのブロッキング呼び出しであってはなりません。では、start メソッド内にエグゼキュータを記述することは良い習慣でしょうか? または、vertx はそのようなケースを自動的に処理しますか。

ここにスニペットがあります 

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");
    }
}

これを非ブロッキング呼び出しにするメカニズムは何ですか。

4

3 に答える 3

1

最も柔軟な方法は、ExecutorService を作成し、それを使用してリクエストを処理することです。これにより、ワーカーのスレッド モデル (固定数または可変数のスレッド、1 つのスレッドでどの作業を連続して実行するかなど) をきめ細かく制御できます。

変更されたサンプルは次のようになります。

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");
    }
}
于 2013-08-21T09:54:05.520 に答える
0

ダフィーモが述べたように、スレッドを作成すると、頂点を使用する目的が無効になります。最善の方法は、メッセージをイベントバスに書き込み、イベントバスからのメッセージをリッスンする新しいハンドラーを作成することです。これを紹介するためにコードを更新しました。「next.topic」トピックにメッセージを書き込み、「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());
       }
    };

}
}
于 2015-07-03T00:18:13.827 に答える