『Javaで共通のサーバプログラムを書く』02リスナー

1つのサーバ・プログラムでは、リスナーの役割は会社のフロントと同様に、ブートの役割を果たすため、リスナーは新しい接続ごとにできるだけ短い時間を費やす必要があります.そうすれば、最も迅速な応答を保証できます.
プログラミング自体に戻ります.
1.Listenerは別々のスレッドで実行することが望ましい
2.リスナーが新しい接続を受けた後、接続を処理する方法はできるだけ早く戻る必要がある
 
Java Push Frameworkでは、通常のクライアントとサーバの監視サービスを同時にリスニングするクライアントが必要なため、AcceptorとMonitorAcceptorの2つのリスナーを定義します.
両者の傍受部に関する論理は同じであるため,まず抽象クラスListenerを定義してモニタの機能を実現し,socketを処理する部分を抽象方法と定義する.

//   socket     

protected abstract boolean handleAcceptedSocket(

            PushClientSocket clientSocket);

 
リスニング機能の実現は簡単ですが、create,bind,acceptの3つのステップです.

    private boolean doListening(InetSocketAddress serverAddr) {

        boolean ret = false;

        int socketBufferSize = getServerImpl().getServerOptions()

                    .getSocketBufferSize();

        int socketType = getServerImpl().getServerOptions()

                    .getSocketType();

        try {

            // Create

            serverSocket = SocketFactory.getDefault().createServerSocket(

                    socketType, socketBufferSize);

            

            // Bind

            serverSocket.bind(serverAddr);



            Debug.debug("Start to listen " + serverAddr.getHostName() + ":"

                    + serverAddr.getPort());

            

            // Accept

            doAccept();

            

            ret = true;

        } catch (IOException e) {

            e.printStackTrace();

            if (serverSocket != null) {

                serverSocket.close();

                serverSocket = null;

            }

        }

        

        return ret;

    }

Javaには現在、同期ブロックSocket、同期非ブロックSocket、JDK 7が新たに追加した非同期Socketのいくつかの異なるsocketがあることを考慮すると、JavaのSocketクラスを直接使用すると、異なるタイプのsocket間で使用を切り替えるのは不便です.PushServer SocketとPushClientSocketの2つの新しいインタフェースをカスタマイズしました.

//      socket  ，       bind accept，

//            close。

public interface PushServerSocket {



    public void bind(InetSocketAddress serverAddr) throws IOException;



    public PushClientSocket accept() throws IOException;



    public void close();

}

//    socket      C++   ，        C++  ，           C++  

public interface PushClientSocket {



    public String getIP();

    public int getPort();

    

    //       Selector       ，       NIO   

    //        

    public SelectionKey registerSelector(Selector selector, int ops, 

            Object attachment) throws IOException;

    

    public int send(byte[] buffer, int offset, int size) throws IOException;

    

    public int recv(byte[] buffer, int offset, int size) throws IOException;

    

    public boolean isOpen();

    

    public boolean isConnected();

    

    public void close();

}

両者に対応するNIOバージョン実装はPushServerSocketImplとPushClientSocketImplであり,コード実装は比較的簡単であるが,ここでは貼らない.
Listenerに戻ってdoAcceptを見てみましょう.

    private void doAccept()

    {

        // Start a new thread

        acceptorThread = new Thread(new Runnable()  {

            public void run() {

                while (blnRunning) {

                    try {

                        PushClientSocket clientSocket = serverSocket.accept();

                        

                        Debug.debug("New client from " + clientSocket.getIP());



                        // Start servicing the client connection

                        if (!handleAcceptedSocket(clientSocket)) {

                            clientSocket.close();

                        }

                    } catch (IOException e) {

                        e.printStackTrace();

                        return;

                    }

                }

            }

            

        });

        

        // Start the thread

        acceptorThread.start();

    }

ここでサーバsocketのacceptメソッド実装はブロックされているため,ポーリングが止まらないことを回避できるため,NIOでacceptを実装する際にconfigureBlockingを非ブロックモードに設定することはできない.
後でリスニングを停止するとサーバsocketのcloseメソッドが直接呼び出され、acceptメソッドは異常を投げ出してループを飛び出し、リスニングスレッドの実行を終了します.
リスニングを終了するときは、スレッドのjoinメソッドを使用してスレッドの終了を待つことを忘れないでください.

    public void stopListening() {

        blnRunning = false;



        // Close server socket

        if (serverSocket != null) {

            serverSocket.close();

            serverSocket = null;

        }

        

        // Wait the thread to terminate

        if (acceptorThread != null) {

            try {

                acceptorThread.join();

            } catch (InterruptedException e) {

                //e.printStackTrace();

            }



            acceptorThread = null;

        }

    }

Acceptorの実装は比較的複雑で、アクセス情報を記録し、いくつかのチェックを行い、ClientFactory処理に渡す必要があります.

    protected boolean handleAcceptedSocket(PushClientSocket clientSocket) {

        //     

        ClientFactory clientFactoryImpl = serverImpl.getClientFactory();

        ServerStats stats = serverImpl.getServerStats();

        ServerOptions options = serverImpl.getServerOptions();



        stats.addToCumul(ServerStats.Measures.VisitorsSYNs, 1);

        //              

        if (clientFactoryImpl.getClientCount() >= options.getMaxConnections()) {

            Debug.debug("Reach maximum clients allowed, deny it");

            return false;

        }



        //  IP     

        if (!clientFactoryImpl.isAddressAllowed(clientSocket.getIP())) {

            Debug.debug("IP refused: " + clientSocket.getIP());

            return false;

        }

        //   socket

        return clientFactoryImpl.createPhysicalConnection(clientSocket, 

                false, listenerOptions);

    }

MonitorAcceptorの実装は比較的簡単で,ClientFactory処理に直接渡せばよい.

    protected boolean handleAcceptedSocket(PushClientSocket clientSocket) {

        return serverImpl.getClientFactory().createPhysicalConnection(

                clientSocket, true, listenerOptions);

    }

ClientFactoryの処理ロジックについては後述する文章で詳しく述べる.
 
リスナー機能を実現するのは簡単なので、言えることは多くありません.