首先，整理NIO进行服务端开发的步骤：
（1）创建ServerSocketChannel,配置它为非阻塞模式。
（2）绑定监听，配置TCP参数，backlog的大小。
（3）创建一个独立的I/O线程，用于轮询多路复用器Selector。
（4）创建Selector,将之前创建的ServerSocketChannel注册到Selector上，监听SelectionKeyACCEPT。
（5）启动I/O线程，在循环体中执行Selector.select()方法，轮训就绪的Channel。
（6）当轮询到了处于就绪状态的Channel时，需要对其进行判断，如果是OP_ACCEPT状态，说明是新的客户端接入，则调用ServerSocketChannel.accept()方法接受新的客户端。
（7）设置新接入的客户端链路SocketChannel为非阻塞模式，配置其他的一些TCP参数。
（8）将SocketChannel注册到Selector,监听OP_READ操作位。
（9）如果轮询的Channel为OP_READ，则说明SocketChannel中有新的就绪的数据包需要读取，则构造ByteBuffer对象，读取数据包。
（10）如果轮询的Channel为OP_WRITE，则说明还有数据没有发送完成，需要继续发送。
Netty时间服务器服务端 TimeServer:
package netty; import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
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; public class TimeServer { public void bind(int port) throws Exception{
//配置服务端的NIO线程组 一个用于服务端接收客户端的连接，另一个用于进行SocketChannel的网络读写
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try{
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup,workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.childHandler(new ChildChannelHandler());
//绑定端口，同步等待成功
ChannelFuture f = b.bind(port).sync();
//等待服务器监听端口关闭
f.channel().closeFuture().sync();
}finally{
//优雅退出，释放线程池资源
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
} private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
protected void initChannel(SocketChannel arg0) throws Exception{
arg0.pipeline().addLast(new TimeServerHandler());
}
}
}
ServerBootstrap是Netty用于启动NIO服务端的辅助类，目的是降低服务端的开发难度。
绑定childChannelHandler,其作用类似于Reactor模式中的handler类，主要用于处理网络I/O事件，例如记录日志、对消息进行编解码等。
使用bind绑定监听端口，随后，调用它的同步阻塞方法sync等待绑定操作完成，完成后Netty会返回一个ChannelFuture,主要用于异步操作的通知回调。
Netty时间服务器服务端 TimeServerHandler:
package netty; import java.io.IOException;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext; public class TimeServerHandler extends ChannelHandlerAdapter{ public void channelRead(ChannelHandlerContext ctx,Object msg) throws IOException{
//将msg转换成Netty的ByteBuf对象
ByteBuf buf = (ByteBuf)msg;
//将缓冲区中的字节数组复制到新建的byte数组中，
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
//获取请求消息
String body = new String(req,"UTF-8");
System.out.println("The time server receive order:" + body);
//如果是"QUERY TIME ORDER"则创建应答消息
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(
System.currentTimeMillis()).toString() : "BAD ORDER";
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
//异步发送应答消息给客户端
ctx.write(resp);
} public void channelReadComplete(ChannelHandlerContext ctx) throws Exception{
ctx.flush();
} public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
ctx.close();
}
}
相比昨天原生的NIO服务端，代码量大大减少。
Netty时间服务器客户端 TimeClient:
package netty; import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
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.NioSocketChannel; public class TimeClient { public void connect(int port,String host) throws Exception{
//创建客户端处理I/O读写的NioEventLoopGroup Group线程组
EventLoopGroup group = new NioEventLoopGroup();
try{
//创建客户端辅助启动类Bootstrap
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>(){
//将ChannelHandler设置到ChannelPipleline中，用于处理网络I/O事件
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new TimeClientHandler());
}
});
//发起异步连接操作，然后调用同步方法等待连接成功。
ChannelFuture f = b.connect(host,port).sync(); //等待客户端链路关闭
f.channel().closeFuture().sync();
}finally{
//优雅退出，释放NIO线程组
group.shutdownGracefully();
}
} public static void main(String[] args) throws Exception{
int port = 8080;
if(args != null && args.length > 0){
try{
port = Integer.valueOf(args[0]);
}catch(NumberFormatException e){
//采用默认值
}
}
new TimeClient().connect(port, "127.0.0.1");
} }
Netty时间服务器客户端 TimeClientHandler:
package netty; import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext; import java.util.logging.Logger; public class TimeClientHandler extends ChannelHandlerAdapter{ private static final Logger logger = Logger.getLogger(TimeClientHandler.class.getName()); private final ByteBuf firstMessage; public TimeClientHandler(){
byte[] req = "QUERY TIME ORDER".getBytes();
firstMessage = Unpooled.buffer(req.length);
firstMessage.writeBytes(req);
} //当客户端与服务端TCP链路简历成功后，Netty的NIO线程会调用该方法，发送查询时间的指令给服务器
public void channelActive(ChannelHandlerContext ctx){
//将请求消息发送给服务端
ctx.writeAndFlush(firstMessage);
} //当服务器返回应答消息时，该方法被调用
public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception{
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String(req,"UTF-8");
System.out.println("Now is :" + body);
} public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){ //释放资源
logger.warning("Unexpected exception from downstream :" + cause.getMessage());
ctx.close();
}
}
运行结果：
Server:
Client:
