search

kmansei / chatapp

0 stars 0 forks source link

Nettyを使用したチャットサーバーコードの解説 #6

Open kmansei opened 1 year ago

kmansei commented 1 year ago

サーバー

package chatapp;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class Server {

    private static final String HOST = "localhost";
    private static final int PORT = 1234;

    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new StringDecoder(), new StringEncoder(), new ServerHandler());
                        }
                    })
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            // ホストアドレスとポート番号を指定してサーバーを起動
            ChannelFuture f = b.bind(HOST, PORT).sync();
            System.out.println("チャットサーバーを起動");

            // ソケットが閉じるまで待機
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

class ServerHandler extends ChannelInboundHandlerAdapter {
    // すべてのアクティブなクライアントを保持するリストまたは集合
    private static final Set<ChannelHandlerContext> channels = ConcurrentHashMap.newKeySet();

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 新しいクライアントが接続した際にリストまたは集合に追加
        channels.add(ctx);
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // クライアントが切断した際にリストまたは集合から削除
        channels.remove(ctx);
        super.channelInactive(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // クライアントから受信したメッセージを出力
        String message = (String) msg;
        System.out.println(message);

        // 他の接続している全てのクライアントにブロードキャスト
        for (ChannelHandlerContext channel : channels) {
            if (channel != ctx) {
                channel.writeAndFlush(message);
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
kmansei commented 1 year ago

Mainメソッド解説

  1. EventLoopGroupの設定

    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

    bossGroup: 新しいクライアント接続を受け入れるためのEventLoopGroup。引数1は、このEventLoopGroupが使用するスレッドの数を示します。 workerGroup: クライアントからのデータ読み取りや書き込みを行うためのEventLoopGroup。デフォルトコンストラクタを使用しているので、利用可能なすべてのCPUコアが利用されます。

  2. ServerBootstrapの設定

    ServerBootstrap b = new ServerBootstrap();
ServerBootstrapオブジェクトを作成。このオブジェクトを使用して、サーバーの設定を行います。

  3. ServerBootstrapのオプション設定

    b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childHandler(new ChannelInitializer<SocketChannel>() { /*...*/ })
.childOption(ChannelOption.SO_KEEPALIVE, true);
group(bossGroup, workerGroup): イベントループグループの設定を行います。
channel(NioServerSocketChannel.class): サーバーが新しい接続を受け入れるために使用するChannelクラスを指定します。
option(ChannelOption.SO_BACKLOG, 128): サーバーソケットが保持できる未処理の接続のキューの最大長を設定します。
childHandler(new ChannelInitializer<SocketChannel>() { /*...*/ }): 新しいクライアント接続ごとに呼び出されるChannelInitializerを設定します。
childOption(ChannelOption.SO_KEEPALIVE, true): すべての子チャンネルに対して、TCP Keepaliveを有効にします。

  4. チャンネルパイプラインの設定

    new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ch.pipeline().addLast(new StringDecoder(), new StringEncoder(), new ServerHandler());
}
}

    ChannelInitializerのinitChannelメソッドで、各新規接続のチャンネルパイプラインを設定します。ここでは、StringDecoder、StringEncoder、ServerHandlerがパイプラインに追加されています。

  5. サーバーのバインド

    ChannelFuture f = b.bind(HOST, PORT).sync();
ServerBootstrapオブジェクトでサーバーをHOSTとPORTにバインドし、バインド操作が完了するまで待ちます。

  6. サーバーの起動確認

    System.out.println("Netty Server has started.");

サーバーが起動したことをコンソールに出力します。

  1. サーバーのクローズ待ち 
    f.channel().closeFuture().sync();

サーバーのチャンネルがクローズされるのを待ちます。

  1. シャットダウン 
    workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();

    最後に、EventLoopGroupをシャットダウンします。これにより、開いているすべてのチャンネルが閉じられ、すべてのスレッドが終了します。

以上が、このmainメソッドのコードの詳細な説明です。

kmansei commented 1 year ago

クラス名.classについて

Javaにおいて、クラス名.classという構文は、指定されたクラスのClassオブジェクトを取得するために使用されます。Classオブジェクトは、リフレクションAPIの一部として、クラスのメタデータ、クラスの構造、クラスに定義されたメソッドやフィールドなどの情報を表します。

使用例 例えば、以下のようにStringクラスのClassオブジェクトを取得できます：

Class<String> stringClass = String.class;
kmansei commented 1 year ago

ChannelOption.SO_BACKLOGとは

ChannelOption.SO_BACKLOG は、Nettyにおけるサーバーソケットのオプションの一つです。このオプションは、システムが接続要求をキューに入れることができる未確認の接続の最大数を指定します。

具体的には、クライアントがサーバーに接続を要求した場合、サーバーはまず接続を「受け入れる」前の一時的なキューにその要求を追加します。このキューは、サーバーがアクティブな接続を処理するのに忙しい場合や、新しい接続を非常に高速に受け入れることができない場合に役立ちます。SO_BACKLOG の値は、このキューのサイズ、つまりキューに入れることができる未確認の接続の最大数を制御します。

以下のように ServerBootstrap の option メソッドを使用して設定することができます：

.option(ChannelOption.SO_BACKLOG, 128)

この例では、サーバーは最大で128の未確認の接続要求をキューに入れることができます。キューが満杯になると、新しい接続要求は拒否されることになります。

この値を適切に設定することは、特にトラフィックの高いサーバーにおいて、パフォーマンスと接続の確立の信頼性のトレードオフを管理する上で重要です。

kmansei commented 1 year ago

ch.pipeline().addLast(...)の部分について

ch.pipeline().addLast(new StringDecoder(), new StringEncoder(), new ServerHandler());

このコードは、Nettyを使用してネットワーク通信を処理するためのパイプライン（Pipeline）を構築しています。Nettyは非常に柔軟なネットワーク通信ライブラリであり、このパイプラインを通じて通信プロトコルを設定し、データのエンコードとデコード、およびデータの処理を行います。

以下は、提供されたコードが行っている具体的な処理です：

ch.pipeline() は、Nettyの Channel オブジェクトに関連付けられたパイプラインを取得します。このパイプラインは、ネットワーク通信の入出力処理を定義するための重要なコンポーネントです。

.addLast(new StringDecoder(), new StringEncoder(), new ServerHandler()) は、パイプラインに3つのハンドラを順番に追加します。

StringDecoder: このハンドラは、バイトデータを文字列にデコードする役割を担います。つまり、クライアントから受信したバイトデータを文字列に変換します。これにより、後続のハンドラで文字列としてデータを扱えます。

StringEncoder: このハンドラは、文字列データをバイトデータにエンコードする役割を担います。つまり、サーバーからクライアントに送信する際に、文字列データをバイトデータに変換します。

ServerHandler: このハンドラは、カスタムのサーバー側の処理を実装します。クライアントから受信したデータを処理し、必要に応じて応答を生成します。このハンドラは、受信したデータの処理やクライアントとの対話をカスタマイズするために使用されます。

このように、パイプラインにはデータのデコードとエンコードを行うハンドラ（StringDecoder と StringEncoder）と、実際のアプリケーションロジックを実装するカスタムハンドラ（ServerHandler）が追加されています。この構成により、クライアントとの通信が効率的に処理され、サーバー側のアプリケーションロジックが適切に実行されます。

kmansei commented 1 year ago

ChannelFutureのsync()とは

ChannelFutureのsync()メソッドは、Nettyというネットワーク通信フレームワークにおいて、非同期操作の終了を待機するためのメソッドです。Nettyは非同期イベントベースのネットワーク通信をサポートしており、ChannelFutureは非同期操作の完了状態を追跡するために使用されます。

具体的には、ChannelFutureのsync()メソッドは以下のように動作します：

ネットワーク通信に関連する非同期操作（例：データの送信、接続の確立、チャンネルのクローズなど）を実行します。これらの操作は通常、非同期に実行されます。

sync()メソッドを呼び出すと、その時点での非同期操作が完了するまで呼び出し元スレッドをブロックします。つまり、非同期操作が終了するのを待機します。

非同期操作が完了した場合、sync()メソッドは処理を再開し、以降のコードを実行します。非同期操作がエラーで終了した場合、例外をスローします。

sync()メソッドは主にネットワーク通信の完了を確認し、次のステップに進む前に操作を同期化するために使用されます。たとえば、データを送信した後にsync()を呼び出すことで、データが正常に送信されたことを確認し、次の操作に進む前に待機することができます。

kmansei commented 1 year ago

childOption(ChannelOption.SO_KEEPALIVE, true);とは

キープアライブの動作： キープアライブが有効になっている場合、TCPは定期的にデータの交換がない接続上にプローブパケットを送信します。 相手側から応答が返ってこない場合、一定時間後に再度プローブパケットを送信し、応答がない場合は接続を閉じます。 これにより、ネットワークの断線や相手側の異常終了等によってデータの交換ができなくなった接続を検出し、リソースの解放やエラー処理を行うことができます。

kmansei commented 1 year ago

なぜServerbootstrapは接続要求とI/O操作を異なるEventLoopGroupに分けているのか

以下chatgptによる考察:

ServerBootstrapで接続要求とI/O操作を分ける理由は、主にパフォーマンスとスケーラビリティの向上のためです。以下に具体的な理由を説明します。

  1. コンカレンシーとスケーラビリティ サーバーが多数のクライアントからの接続要求を効率良く処理するためには、接続要求の受付とデータの読み書きを並行して処理することが重要です。 bossGroupは接続要求の受付のみを行い、接続が確立されたらそのチャンネルをworkerGroupに渡します。 workerGroupは接続されたチャンネルからのデータの読み取りや書き込みを並行して処理します。 これにより、大量の接続要求とデータトランスファーを効率良く、スケーラブルに処理することができます。
  2. リソースの効率的な利用 接続の受付とデータの読み書きは、リソース利用の観点からも異なる性質を持っています。 接続要求は短時間で処理できるため、一つのEventLoopで多くの接続要求を処理できます。 一方で、データの読み書きは、通信量や処理内容によっては、より多くの時間とリソースを消費する可能性があります。 したがって、これらを分けて扱うことで、各種リソースをより効率的に利用できます。
  3. 負荷の分散 bossGroupとworkerGroupを分けることで、異なるCPUコアやスレッド上で並行にタスクを実行でき、システムの負荷を分散させることができます。 これにより、一つのタスクが他のタスクのパフォーマンスに影響を与えにくくなります。 このように、ServerBootstrapでは、接続要求の処理とI/O操作の処理を分けることで、システム全体のパフォーマンスやスケーラビリティを向上させています。