Netty 权威指南笔记（二）：Java NIO 和 Netty 对比
Netty 是业界流行的 NIO 框架之一，它的健壮性、功能、性能、可定制性和可扩展性在同类框架中都说首屈一指的，也已经得到了成百上千商用项目的验证。Netty 框架都有什么优点呢？
API 使用简单，开发门槛低。
功能强大，预置多种编解码功能，支持多种主流协议。
定制能力强，可以通过 ChannelHandler 对通信框架灵活扩展。
性能高。
成熟稳定，社区活跃，已经修复了 Java NIO 所有的 Bug。
经历了大规模商业应用的考验，质量有保证。
Java NIO 开发
我们看一看在 笔记（一）里面的 Java NIO 示例程序，回顾一下使用 NIO 开发服务端程序的步骤：
创建 ServerSocketChannel 和业务处理线程池。
绑定监听端口，并配置为非阻塞模式。
创建 Selector，将之前创建的 ServerSocketChannel 注册到 Selector 上，监听 SelectionKey.OP_ACCEPT。
循环执行 Selector.select() 方法，轮询就绪的 Channel。
轮询就绪的 Channel 时，如果是处于 OP_ACCEPT 状态，说明是新的客户端接入，调用 ServerSocketChannel.accept 接收新的客户端。
设置新接入的 SocketChannel 为非阻塞模式，并注册到 Selector 上，监听 OP_READ。
如果轮询的 Channel 状态是 OP_READ，说明有新的就绪数据包需要读取，则构造 ByteBuffer 对象，读取数据。
Netty 开发
在讲 Netty 优点的时候，说到“Netty API 使用简单，开发门槛低”。先看看下面的 Netty TimeServer 示例程序，看看 Netty 开发都有哪些步骤？
创建 NIO 线程组 EventLoopGroup 和 ServerBootstrap。
设置 ServerBootstrap 的属性：线程组、SO_BACKLOG 选项，设置 NioServerSocketChannel 为 Channel，设置业务处理 Handler。
绑定端口，启动服务器程序。
在业务处理 TimeServerHandler 中，读取客户端发送的数据，并给出响应。
那么相比 Java NIO，使用 Netty 开发程序，都简化了哪些步骤呢？
OP_ACCEPT 的处理被简化，因为对于 accept 操作的处理在不同业务上都是一致的。
在 NIO 中需要自己构建 ByteBuffer 从 Channel 中读取数据，而 Netty 中数据是直接读取完成存放在 ByteBuf 中的。相当于省略了用户进程从内核中复制数据的过程。
在 Netty 中，我们看到有使用一个解码器 FixedLengthFrameDecoder，可以用于处理定长消息的问题，能够解决 TCP 粘包读半包问题，十分方便。
示例程序
服务器端程序
public class TimeServer {
public void bind(int port) throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 2014)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(16));
socketChannel.pipeline().addLast(new TimeServerHandler());
}
});
ChannelFuture future = bootstrap.bind(port).sync();
System.out.println("start listening ...");
future.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
int port = 11000;
new TimeServer().bind(port);
}
}
TimeServerHandler 业务处理类：
public class TimeServerHandler extends ChannelHandlerAdapter {
private int counter = 0;
@Override
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("time server receive: " + body + ", counter: " + counter++);
String currentTime = "QUERY TIME ORDER".equals(body) ? new Date().toString() : "BADY ORDER";
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
ctx.write(resp);
}
/**
* 读完一批次数据，就会调用一次 channelReadComplete
* @param ctx
* @throws Exception
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
System.out.println("channelReadComplete");
// flush 之后，之前 write 的数据，才会发送出去
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("exceptionCaught");
cause.printStackTrace();
ctx.close();
}
}
客户端程序
public class TimeClient {
public void connect(int port, String host) throws InterruptedException {
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try {
List<VirtualMachineDescriptor> list = VirtualMachine.list();
for (VirtualMachineDescriptor descriptor : list) {
System.out.println(descriptor.displayName());
}
System.out.println("vm: " + list.size());
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(28));
socketChannel.pipeline().addLast(new TimeClientHandler());
}
});
ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
channelFuture.channel().closeFuture().sync();
System.out.println("end listening ...");
} finally {
eventLoopGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
int port = 11000;
new TimeClient().connect(port, "127.0.0.1");
}
}
TimeClientHandler 客户端业务处理类：
public class TimeClientHandler extends ChannelHandlerAdapter {
private final byte[] req;
private int counter = 0;
public TimeClientHandler() {
System.out.println("init TimeClientHandler");
req = "QUERY TIME ORDER".getBytes();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for (int i = 0; i < 100; ++i) {
ByteBuf firstMessage = Unpooled.buffer(req.length);
firstMessage.writeBytes(req);
ctx.writeAndFlush(firstMessage);
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("channelRead");
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
System.out.println("now is: " + new String(req, "UTF-8") + ", counter: " + counter++);
if (counter > 99) {
ctx.close();
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("exceptionCaught");
ctx.close();
}
|
|
收藏
淘帖
踩
