永続的な接続をサポートするサーバー側のJavaコードコンテナー

免責事項

この記事に記載されているものはすべて個人的な実務経験であり、「究極の真実」のタイトルであると主張するものではありません。

前文

こんにちは。 コンピューターゲームを作ることに興味があります。 私が常に新しいことを改善し、学ぼうとしている私のお気に入りの方向性は、ブラウザベースのマルチプレイヤーゲームです。

Apache Tomcatは、1つのアイデアのプロトタイプを作成するためのサーブレットコンテナーとして使用されます。 彼は、httpプロトコルを介してクライアント部分と通信します。 このタイプのゲームでは、スキームは非常に有効であり、実装は非常に簡単です。

しかし、時期尚早な最適化の1つ（はい、これは悪いですが、ここでは許可することにしました）は、サーバーとクライアントの間に永続的な接続を使用するというアイデアでした。 このようなスキームでは、各リクエストで接続を開いたり閉じたりする時間は無駄になりません。 このスキームを実装するために、TomcatのWebSocket APIを検討しましたが、自転車を書くのが面白くなったので、猫の下での開発に関する話に出会ってください。

ツール

そのため、このアイデアを実装するために、次を使用しました。

• NetBeans 7.2.1：実際には、この問題を解決するためにすべてのJavaコードが記述されています
• Jdk 1.7
• Netty：多数の接続でサーバーができる限り生産的になるようにnioを使用することが決定され、このフレームワークは正常に機能しました。
• ソケットIO：クライアント側
• Apache Tomcat 7.0.27：ベンチマーク用
• Maven：これらすべてを構築する

建築

まず、アプリケーションロジックを検討します。



コンテナは、メインクラスSocketServletContainerで表されます。 コンテナを起動/停止するのに役立ち、サーブレットを管理するメソッドも含まれています。 この記事では、サーブレットという用語は、サーバーコードを持つメソッドを含むオブジェクトを指し、JCPサーブレット仕様とは関係がないことに注意してください。 このようなオブジェクトサーブレットを呼び出す方が便利です。



実際には、すべてのユーザーサーブレットが継承されるベースクラスServletと 、セッションに関する情報を保存し、ユーザーにメッセージを送信するために使用される接続セッションクラス（ SocketSession ）があります（その理由は後で説明します）。 着信および発信バッファのクラス（ InputBufferおよびOutputBuffer ）もそれぞれ実装されました。



また、xml形式の構成ファイルを解析するヘルパークラスConfigを実装する必要がありました。 QueueHandlerクラスとTaskHandlerクラスについて言及する必要があります 。



QueueHandlerは要求キューハンドラーであり、処理するTaskクラスのインスタンスを追加するメソッドが含まれています。

TaskHandlerはRunnableインターフェースを実装します。 runメソッドには、送信されたリクエストの処理が含まれます。

Taskクラスには、到着したリクエストに関する情報（サーバーに渡されるパラメーターが参照するサーブレット）およびネットワークを操作するためのメソッド（読み取り\書き込み）が含まれます。



次に、ネットワークを使用した作業の編成を検討します。



Nettyでの作業については詳しく説明しません。Nettyに関する記事についてはRena4ka habrayuzerに感謝します）。 公式サイトのhabrまたはドキュメントを読むと、より便利になります。 Nettyのプログラミング経験のない人が基本原則を理解するために必要な部分のみを検討します。

ServerPipelineFactoryクラスはChannelPipelineファクトリーであり、Nettyが機能するために必要です。 Decoder 、 Encoder 、 NioHandlerの 3つのクラスも実装する必要がありました 。

最初の2つは、サーバーに到着したパケットのハンドラーです。 デコーダーは、ネットワークからのパケットを正しく解析し、 Taskクラスのインスタンスを返します。 エンコーダーは、 タスクインスタンスをネットワークに正しく書き込み、クライアントに送信する責任があります。

実際、 NioHandlerはネットワークマネージャーです。接続を受け入れ、処理のためのタスクを送信し、セッションを管理します。

プロトコル

クライアントとサーバー間で通信するには、独自のプロトコルが必要です。 私はそれをかなりシンプルでテキストにすることを決めました。

その結果、クライアントは、servlet_name [sysDiv] request_parametersのようなクエリ文字列をサーバーに送信します。

クエリパラメータのリストの形式：name1 = value1、name2 = value2、...



例 ：「TS [sysDiv]メッセージ= Hello habrahabr.ru」。



クライアントは、応答を生成したサーブレットと渡されたパラメーターのリストを示す文字列を受け取るという意味で、プロトコルは対称的であることに注意してください。



そして、コンテナのコー​​ドの検討に直接渡します。 しかし、まず最初に。

構成ファイルの形式

<?xml version="1.0" encoding="utf-8"?> <config> <address>localhost</address> <port>9999</port> <workThreadCount>2</workThreadCount> <processThreadCount>2</processThreadCount> </config>
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

workThreadCount-ネットワークからメッセージを受信し、ネットワークに書き込むスレッドの数（Nettyを初期化するために必要）。

processThreadCount-サーバーに到着したリクエストの一般的なキューを処理するスレッドの数。 実際には、クエリ文字列の解析、すべてのサーバーコードの操作、および応答の形成があります。

SocketServletContainer

このクラスは「中心」クラスであり、プログラムの他のクラスからアクセスする方が便利なため、シングルトンです。 そして、もちろん、アプリケーションごとにサーバーのコピーが1つ含まれていることを意味します（したがって、スレッドセーフなシングルトン実装は必要ありません）。 私の意見では、これは論理的です。

 public class SocketServletContainer { private Channel channel; private ServerBootstrap networkServer; private QueueHandler queueHander; private Map<String, Servlet> servlets; private Config conf; private static SocketServletContainer server= null; private static List<SocketSession> list= new ArrayList<SocketSession>(); public List<SocketSession> getListSession() { return list; } static public SocketServletContainer getInstance() { if (server==null) { server= new SocketServletContainer(); } return server; } private SocketServletContainer() { conf= new Config("conf.xml"); // ,   -  Exception. try { conf.read(); } catch(Exception e) { throw new ContainerInitializeException(e.toString()); } servlets= new HashMap<String, Servlet>(); } public void start() { // Netty ExecutorService bossExec = new OrderedMemoryAwareThreadPoolExecutor(1, 400000000, 2000000000, 60, TimeUnit.SECONDS); ExecutorService ioExec = new OrderedMemoryAwareThreadPoolExecutor(conf.getWorkThreadCount(), 400000000, 2000000000, 60, TimeUnit.SECONDS); networkServer = new ServerBootstrap(new NioServerSocketChannelFactory(bossExec, ioExec, conf.getWorkThreadCount())); networkServer.setOption("backlog", 500); networkServer.setOption("connectTimeoutMillis", 10000); networkServer.setPipelineFactory(new ServerPipelineFactory()); channel = networkServer.bind(new InetSocketAddress(conf.getAddress(), conf.getPort())); //    queueHander= new QueueHandler(conf.getProcessThreadCount()); System.out.println("Ready"); } // «»   public void stop() { if (channel.isOpen()) { ChannelFuture future= channel.close(); future.awaitUninterruptibly(); } queueHander.stop(); } public QueueHandler getQueueHandler() { return this.queueHander; } //      public void registerServlet(Servlet servlet, String name) { //    -    HashMap. synchronized(servlets) { if (!servlets.containsKey(name)) { servlets.put(name, servlet); } } } public Servlet getServlet(String name) { return servlets.get(name); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

サーブレット

ここではすべてがシンプルで明確です。 doRequestメソッドは、このサーブレットの呼び出しを指示するパケットが到着すると呼び出されます。

Zam ：doRequestメソッドにセッションを渡すことは、サーブレットが利用可能なすべてのセッションのリストを取得してメッセージを送信できるようにするために行われます。 たとえば、チャットを実装する場合。

 abstract public class Servlet { abstract public void doRequest(InputBuffer input, OutputBuffer output, SocketSession session); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

SocketSession

各セッションには固有のIDがあります。 接続された20,000のクライアントのIDニックネームのプールがあります。 この制限を超えると、サーバーはセッションを作成しようとするときにエラーを記録し、非クライアントエラーメッセージを送信してチャネルを閉じます。

プールサイズの値は経験的に計算する方が適切です。理想的には、サーバー上で同時に接続可能なクライアントの最大数よりもわずかに大きくする必要があります。

 public class SocketSession { private static byte[] idPool; public int generateId() { synchronized(idPool) { if (idPool==null) { idPool= new byte[20000]; for (int j=0;j<idPool.length;j++) { idPool[j]=0; } } for (int j=0;j<idPool.length;j++) { if (idPool[j]==0) { idPool[j]=1; return j; } } return -1; } } private int id; private Channel channel; //      List  . public SocketSession(Channel channel) { this.channel= channel; this.id= generateId(); //    if (this.id==-1) { OutputBuffer out= new OutputBuffer(); out.setPar("error", "Connection limit error"); send(out, "System Servlet"); //  System.err.println("Connection limit error"); return; } SocketServletContainer.getInstance().getListSession().add(this); } public int getId() { return id; } //  .            . public void send(OutputBuffer output, String servletName) { synchronized(channel) { channel.write(new Task(servletName, output.toString())); } } // , ,      - «»  public void close() { synchronized(idPool) { idPool[this.id]= 0; } channel.close(); SocketServletContainer.getInstance().getListSession().remove(this); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

入力バッファ

初期化はコンストラクターで行われ、ソース文字列には、指定された形式のクエリパラメーターのリストが含まれている必要があります。

 public class InputBuffer { private Map<String, String> map= new HashMap<String, String>(); public InputBuffer(String source) { String[] par= source.split(","); for (int j=0; j< par.length; j++) { if (!par[j].contains("=")) { continue; } String[] data= par[j].split("="); if (data.length<2) { System.err.println("Parsing Error"); continue; } map.put(data[0], data[1]); } } public String getPar(String key) { return map.get(key); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

OutputBuffer

クラスインターフェイスは非常に簡単です。 重要な注意点は、toString（）メソッドをオーバーライドする必要があることです。これは、 SocketSessionクラスで応答を形成するために使用されるためです。

 public class OutputBuffer { private List<String> list= new ArrayList<String>(); public void setPar(String key, String par) { list.add(key+"="+par); } @Override public String toString() { StringBuilder res= new StringBuilder(); for (int j=0; j< list.size();j++) { res.append(list.get(j)); if (j!=list.size()-1) { res.append(","); } } return res.toString(); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

構成

SocketServletContainerで使用されているものからインターフェイスが明確であり、インターネット上のJavaにはxmlパーサーの実装が多数あるため、このクラスの実装は提供しません。

個人的には、DOMパーサーを使用しました。

キューハンドラー

このクラスは実装も非常に簡単です。 内部には、タスクを処理するスレッドプール（ TaskHandler ）が含まれています。 私は、計画を信頼できる実証済みのthreadPool実装に移行しました。 Executors.newFixedThreadPool（n）ファクトリーを使用してプールを作成します。



stopメソッドが呼び出されると、キュー内の既存のタスクは処理されますが、新しいTaskHandlersは処理されません。

 public class QueueHandler { private ExecutorService threadPool; private int threadPoolSize; public QueueHandler(int size) { threadPoolSize= size; threadPool= Executors.newFixedThreadPool(threadPoolSize); } public void stop() { threadPool.shutdown(); } public void addTaskToProcess(Task task, SocketSession session) { threadPool.execute(new TaskHandler(task, session)); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

タスクハンドラー

ここでもすべてが非常に簡単です。 プレーヤーのセッションと処理されるタスクは、コンストラクターに転送されます。

 public class TaskHandler implements Runnable{ private Task task; private SocketSession session; public TaskHandler(Task task, SocketSession session) { this.task= task; this.session= session; } @Override public void run() { //     Servlet servlet= SocketServletContainer.getInstance().getServlet(task.getServletName()); OutputBuffer output= new OutputBuffer(); //  ,   . if (servlet==null) { output.setPar("error", "servlet not found"); session.send(output, "Error Message"); return; } //     servlet.doRequest(new InputBuffer(task.getBuffer()),output, session); //  . session.send(output, task.getServletName()); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

タスク

Taskオブジェクトには、「servlet name」フィールドと「buffer」フィールドがあります。 バッファはクエリパラメータの文字列です。

Nettyが機能するには、クラスのインスタンスを取得したり、パイプに書き込んだりするために、静的な書き込み/読み取りメソッドが必要です。

 public class Task { private String servletName=""; private String buffer=""; public Task(String servletName, String buffer) { this.servletName= servletName; this.buffer= buffer; } public Task() { } public String getServletName() { return servletName; } public String getBuffer() { return buffer; } //    public void get(ChannelBuffer buffer) { int length= buffer.readInt(); byte[] bytes= new byte[length]; buffer.readBytes(bytes); String input= new String(bytes); String[] data= input.split(java.util.regex.Pattern.quote("[sysDiv]")); if (data.length<2) { System.err.println("Parsing error"); return; } this.servletName= data[0]; this.buffer= data[1]; } //    public void send(ChannelBuffer buffer) { String output= this.servletName + "[sysDiv]"+ this.buffer; buffer.writeInt(output.getBytes().length); buffer.writeBytes(output.getBytes()); } public static Task read(ChannelBuffer buffer) { Task task= new Task(); task.get(buffer); return task; } public static void write(Task task, ChannelBuffer buffer) { task.send(buffer); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ネットワーク部

約束どおり、私はnettyを詳細に操作することを検討しません。コードを提供し、ロジックの実装に関連するポイントを説明するだけです。

ServerPipelineFactory

 public class ServerPipelineFactory implements ChannelPipelineFactory { @Override public ChannelPipeline getPipeline() throws Exception { return Channels.pipeline(new Encoder(),new Decoder(),new NioHandler()); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

デコーダー

パケットは、次の形式でサーバーに到着します。最初の4バイトは「有用な」データの長さ、次にデータ自体です。 デコーダーは読み取りを実装しているため、上のレイヤーでは、データがまだ完全に到着していないという事実を考えることはできません。

 public class Decoder extends ReplayingDecoder<DecoderState> { public enum DecoderState { READ_LENGTH, READ_CONTENT; } public Decoder() { super(DecoderState.READ_LENGTH); } private int length; @Override public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { ctx.sendUpstream(e); } @Override public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { ctx.sendUpstream(e); } @Override protected Object decode(ChannelHandlerContext arg0, Channel arg1, ChannelBuffer buffer, DecoderState state) { switch (state) { case READ_LENGTH: length = buffer.readInt(); checkpoint(DecoderState.READ_CONTENT); case READ_CONTENT: ChannelBuffer frame= buffer.readBytes(length); // task       Task task= Task.read(frame); checkpoint(DecoderState.READ_LENGTH); return task; default: throw new Error( "Shouldn't reach here" ); } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

エンコーダー

 public class Encoder extends OneToOneEncoder { @Override protected Object encode(ChannelHandlerContext channelhandlercontext, Channel channel, Object obj) throws Exception { //    Task,      if(!(obj instanceof Task)) { return obj; } Task task= (Task)obj; ChannelBuffer buffer = ChannelBuffers.dynamicBuffer(); // task   Task.write(task, buffer); return buffer; } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ニオハンドラー

このオブジェクトは、ネットワークの操作、クライアントの接続、メッセージの受信、切断などの主なイベントを処理します。

 public class NioHandler extends SimpleChannelUpstreamHandler { private SocketSession session; @Override public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { //     session= new SocketSession(e.getChannel()); System.out.println("Has connect"); } @Override public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { session.close(); } @Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) { if(e.getChannel().isOpen()) { //  Task       QueueHandler. Task message= (Task)e.getMessage(); SocketServletContainer.getInstance().getQueueHandler().addTaskToProcess(message, session); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) { //  .  ,    . session.close(); e.getCause().printStackTrace(System.err); ctx.getChannel().close(); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

サーブレットの例

 public class TS extends Servlet { @Override public void doRequest(InputBuffer input, OutputBuffer output, SocketSession session) { output.setPar("request", input.getPar("message")+session.getId()); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

仕組みまたはアプリケーションのメインクラス

実際、アプリケーションにはそれほど多くのコード行がなく、すべてがシンプルで透過的です。

 public class App { public static void main( String[] args ) throws ContainerInitializeException { SocketServletContainer server= SocketServletContainer.getInstance(); server.registerServlet(new TS(), "TS"); server.start(); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

いくつかのテスト

さて、コンテナは書かれており、動作します。 私はそれのためのクライアントラッパーを作成することを気にしませんでした、私はソケットに直接書き込むことに自分自身を制限しました、それは次のように見えます：

 socket = new Socket("127.0.0.1", 9999); DataOutputStream dos= new DataOutputStream(socket.getOutputStream()); DataInputStream dis= new DataInputStream(socket.getInputStream()); String buffer= "TS[sysDiv]message=IloveJava"; dos.writeInt(buffer.getBytes().length+4); dos.writeInt(buffer.getBytes().length); dos.write(buffer.getBytes()); dos.flush();
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

一般に、記事の冒頭で述べたように、そのようなシステムの作成は、私が自分自身に許可することにした時期尚早な最適化の1つです。 したがって、私たち全員が書いたので、いくつかのテストを行わないのは愚かなことです。



実際、このソリューションをhttpで実行されているサーブレットコンテナと比較することにしました。

テストのために、Tomcatでスピンするサーブレットと、作成されたコンテナー内で動作するサーブレットが開発されました。



Zam： Tomcatが正常にサポートしたWebソケットは、このゲームプロジェクトの実装では考慮されていなかったため、httpプロトコルのパフォーマンスとソケットのソリューションを意図的に比較しました。



テストの特徴：

• 両方のサーブレットは、ほぼ同じ操作を実行しました。つまり、同じ行を出力ストリームに書き込みました。
• 興味のあるテストパラメーターは、要求を処理するときのサーバーからの応答時間です
• 測定はローカルホストで行われました
• 標準のJava言語ツールを使用して応答時間を得ました
• 特定の数値には興味がありませんでしたが、テストは非常に「大まかに」書かれているため、結果の順序を比較することに興味がありました。
• 各テストで、同じパラメーターを使用して10,000件のクエリを実行し、その後平均値を計算しました

結果は次のとおりです。平均して、Tomcatの1つの「空の」サーブレットの処理には0、99ミリ秒かかりました。

この記事で説明されているコンテナは、 0.09ミリ秒で同様のタスクに対処しました。



桁違いの2つの結果があります。 しかし、ソケットを使用するというアイデアは、速度のためではなく、サーバーからクライアントに手を差し伸べることができる必要があるため、私に思いついたという事実を考慮すると、結果は満足のいくものと考えることができます。

TODO：

同様のシステムに実装できる短いリストもありました。

1. 入力データの検証。 入力バッファーに検証（文字列マスク）メソッドを追加すると、対応するパラメーターのデータ型マスクを使用して、文字列だけでなく目的の型に自動的に変換されます。 次のようになります：validate（ "message：String、count：int"）;
2. データ暗号化を追加します。 プロトコルはテキストですが、バイト[]バッファーが書き込まれるのはwriteUTF8（）ではなく、このためです。 インターフェースCrypto {}を実装できます。これには、code（）およびencode（）の2つのメソッドがあります。 また、暗号化アルゴリズムを簡単に変更または選択できるように、このようなインターフェイスの実装はSocketServletContainer（）に渡す必要があります。
3. （Tomcatで行われているように）アノテーションを処理し、サーブレットの初期化を遅らせます。
4. デリミタシールドを使用した入力バッファのより安全な解析
5. そのようなシステムで必要になるかもしれない他の便利な小さなものの束。

結論の代わりに

コンテナの結果は私の期待を完全に満たしました。 NIOを使用することで、フローを経済的に消費し、既存のハードウェア用のコンテナを構成して、可能な限り効率的に動作することができました。

コンテナが提供する機能により、パケット解析などの「低レベル」なことを心配せずにアプリケーションを簡単に開発できます（Tomcatでの開発に使用されているように思われるものすべて:)）。



しかし、実際には、このようなソリューションを実際のプロジェクトの基礎として使用することを敢えてしませんでした。なぜなら、私にとっては、もちろんソケットの存在は非常に有用だからです（サーバーとクライアントのフィードバックにとって）が、原則として、それは重要ではありません しかし、長年にわたって実証されたTomcatのパフォーマンスと信頼性、そして何千人もの開発者が疑問を投げかけることはありません。

私は、実装されたシステムを「狭い」が、httpプロトコルを使用するのがよくない、重要ではない場所で使用する予定です。たとえば、チャットの実装に最適です。



この記事が読者にとって興味深いものであり、誰かにとって役立つことを願っています。 私は全体像を伝えようとしましたが、たぶん、推論とコードの量が多すぎました。 文書に関する質問や提案を聞いてうれしく思います。