Nettyの基本的な使用方法を示すために、1つのサーバーと5つのクライアントスレッド間の通信を実装します。クライアントはサーバーに接続後メッセージを送信し、サーバーは受信したメッセージに対して応答を返却します。クライアントは応答を受信後、ランダムな時間待機して再びメッセージ送信を繰り返します。
プロジェクト構成：
サーバー実装:
package Server;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.*;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.execution.ExecutionHandler;
import org.jboss.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor;
import Util.Constant;
public class NettyServer {
public static String ホストアドレス = "127.0.0.1";
// スレッドプールの生成
static ExecutionHandler 実行ハンドラ = new ExecutionHandler(new OrderedMemoryAwareThreadPoolExecutor(16, 1048576, 1048576));
public static void main(String[] args) {
// チャンネルファクトリの生成
final ChannelFactory チャンネルファクトリ = new NioServerSocketChannelFactory(
// 接続処理用スレッドプール
Executors.newCachedThreadPool(),
// NIOにより1つのワーカースレッドで複数チャネルを管理可能
Executors.newCachedThreadPool());
// サーバー設定
ServerBootstrap ブートストラップ = new ServerBootstrap(チャンネルファクトリ);
ServerPipelineFactory パイプラインファクトリ = new ServerPipelineFactory(実行ハンドラ);
ブートストラップ.setPipelineFactory(パイプラインファクトリ);
// Nagleアルゴリズムの無効化
ブートストラップ.setOption("child.tcpNoDelay", true);
// TCPキープアライブの有効化
ブートストラップ.setOption("child.keepAlive", true);
// 複数ポートの監視
ブートストラップ.bind(new InetSocketAddress(Constant.p1));
System.out.println("ポート " + Constant.p1 + " を監視中...");
ブートストラップ.bind(new InetSocketAddress(Constant.p2));
System.out.println("ポート " + Constant.p2 + " を監視中...");
ブートストラップ.bind(new InetSocketAddress(Constant.p3));
System.out.println("ポート " + Constant.p3 + " を監視中...");
ブートストラップ.bind(new InetSocketAddress(Constant.p4));
System.out.println("ポート " + Constant.p4 + " を監視中...");
ブートストラップ.bind(new InetSocketAddress(Constant.p5));
System.out.println("ポート " + Constant.p5 + " を監視中...");
}
}
パイプラインファクトリ:
package Server;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;
import org.jboss.netty.handler.execution.ExecutionHandler;
public class ServerPipelineFactory implements ChannelPipelineFactory {
private final ExecutionHandler 実行ハンドラ;
public ServerPipelineFactory(ExecutionHandler 実行ハンドラ){
this.実行ハンドラ = 実行ハンドラ;
}
@Override
public ChannelPipeline getPipeline() throws Exception {
return Channels.pipeline(
new StringEncoder(),
new StringDecoder(),
// 複数パイプライン間で共有する実行ハンドラを配置
実行ハンドラ,
// ビジネスロジックハンドラ
new MyServerHandler());
}
}
サーバーハンドラ:
package Server;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import Util.Tool;
public class MyServerHandler extends SimpleChannelHandler{
@SuppressWarnings("static-access")
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
System.out.println("サーバー受信:" + e.getMessage());
// ランダム秒数の待機後送信
Thread th = Thread.currentThread();
int 間隔 = Tool.getInterval(100);
th.sleep(間隔*1000);
e.getChannel().write("サーバーからの応答: こんにちは！");
super.messageReceived(ctx, e);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
e.getCause().printStackTrace();
Channel ch = e.getChannel();
ch.close();
super.exceptionCaught(ctx, e);
}
@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
System.out.println("クライアント接続！");
super.channelConnected(ctx, e);
}
}
クライアント実装:
package Client;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.*;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.execution.ExecutionHandler;
import org.jboss.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor;
import Util.Constant;
public class NettyClient extends Thread{
public static String ホストアドレス = "127.0.0.1";
ClientBootstrap ブートストラップ;
int 接続先ポート;
// スレッドプールの生成
static ExecutionHandler 実行ハンドラ = new ExecutionHandler(new OrderedMemoryAwareThreadPoolExecutor(16, 1048576, 1048576));
public NettyClient(int 接続先ポート) {
this.接続先ポート = 接続先ポート;
// チャンネルファクトリの生成
final ChannelFactory チャンネルファクトリ = new NioClientSocketChannelFactory(
// 接続処理用スレッドプール
Executors.newCachedThreadPool(),
// ワーカースレッドプール
Executors.newCachedThreadPool());
// クライアント設定
ブートストラップ = new ClientBootstrap(チャンネルファクトリ);
ClientPipelineFactory パイプラインファクトリ = new ClientPipelineFactory(実行ハンドラ);
ブートストラップ.setPipelineFactory(パイプラインファクトリ);
ブートストラップ.setOption("tcpNoDelay" ,true);
ブートストラップ.setOption("keepAlive", true);
ブートストラップ.connect(new InetSocketAddress(接続先ポート));
}
public void run(){
ChannelFuture フューチャー = ブートストラップ.connect(new InetSocketAddress(ホストアドレス, 接続先ポート));
// 接続開始
System.out.println(接続先ポート + " への接続を開始...");
// 終了まで待機
フューチャー.getChannel().getCloseFuture().awaitUninterruptibly();
// リソース解放
ブートストラップ.releaseExternalResources();
}
public static void main(String[] args) {
NettyClient nc1 = new NettyClient(Constant.p1);
NettyClient nc2 = new NettyClient(Constant.p2);
NettyClient nc3 = new NettyClient(Constant.p3);
NettyClient nc4 = new NettyClient(Constant.p4);
NettyClient nc5 = new NettyClient(Constant.p5);
nc1.start();
nc2.start();
nc3.start();
nc4.start();
nc5.start();
}
}
クライアントパイプラインファクトリ:
package Client;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;
import org.jboss.netty.handler.execution.ExecutionHandler;
public class ClientPipelineFactory implements ChannelPipelineFactory {
private final ExecutionHandler 実行ハンドラ;
public ClientPipelineFactory(ExecutionHandler 実行ハンドラ){
this.実行ハンドラ = 実行ハンドラ;
}
@Override
public ChannelPipeline getPipeline() throws Exception {
return Channels.pipeline(
new StringEncoder(),
new StringDecoder(),
// 複数パイプライン間で共有する実行ハンドラを配置
実行ハンドラ,
// ビジネスロジックハンドラ
new MyClientHandler());
}
}
クライアントハンドラ:
package Client;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import Util.Tool;
public class MyClientHandler extends SimpleChannelHandler{
// サーバー接続時のメッセージ送信
@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
System.out.println("サーバーに接続しました！");
e.getChannel().write("クライアントからの挨拶: こんにちは！ " + System.currentTimeMillis());
super.channelConnected(ctx, e);
}
@SuppressWarnings("static-access")
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
System.out.println("クライアント受信:" + e.getMessage());
// ランダム秒数の待機後送信
Thread th = Thread.currentThread();
int 間隔 = Tool.getInterval(5);
th.sleep(間隔*1000);
e.getChannel().write("クライアントからの挨拶: こんにちは！ " + System.currentTimeMillis());
super.messageReceived(ctx, e);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
e.getCause().printStackTrace();
Channel ch = e.getChannel();
ch.close();
super.exceptionCaught(ctx, e);
}
}
定数クラス:
package Util;
public class Constant {
final static int ベースポート = 10000;
public static int p1 = ベースポート + 1;
public static int p2 = ベースポート + 2;
public static int p3 = ベースポート + 3;
public static int p4 = ベースポート + 4;
public static int p5 = ベースポート + 5;
}
ユーティリティ:
package Util;
import java.util.Random;
public class Tool {
static Random ランダムジェネレータ = new Random();
public static int getInterval(int 最大値){
return ランダムジェネレータ.nextInt(最大値);
}
}