Java网络编程 — Netty入门

认识Netty

Netty简介

Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients.

Netty is a NIO client server framework which enables quick and easy development of network applications such as protocol servers and clients. It greatly simplifies and streamlines network programming such as TCP and UDP socket server.

Netty是一个异步事件驱动的网络应用框架，用于快速开发可维护的高性能协议服务器和客户端。

Netty是一款NIO客户端服务器框架，可以快速轻松地开发协议服务器和客户端等网络应用程序。它极大地简化并简化了TCP和UDP套接字服务器等网络编程。

Netty架构图如下图所示：

Netty的特性总结如下表：

为什么选择Netty

Netty是业界最流行的NIO框架之一，它的健壮性、功能、性能、可定制性和可扩展性在同类框架中都是首屈一指的，已经得到成百上千的商用项目验证。通过对Netty的分析，将它的优点总结如下：

API使用简单，开发门槛低；
功能强大，预置了多种编解码功能，支持多种主流协议；
定制能力强，可以通过ChannelHandler对通信框架进行灵活地扩展；
性能高，通过与其他业界主流的NIO框架对比，Netty的综合性能最优；
成熟、稳定，Netty修复了已经发现的所有JDK NIO BUG，业务开发人员不需要再为NIO的BUG而烦恼；
社区活跃，版本迭代周期短，发现的BUG可以被及时修复，同时，更多的新功能会加入；
经历了大规模的商业应用考验，质量得到验证。

核心概念

Netty is a non-blocking framework. This leads to high throughput compared to blocking IO. Understanding non-blocking IO is crucial to understanding Netty’s core components and their relationships.

Channel

Channel is the base of Java NIO. It represents an open connection which is capable of IO operations such as reading and writing.

Channel是Java NIO的基础。它表示一个开放的连接，能够进行IO操作，例如读写。

Future

Netty中的每个IO操作都是非阻塞的，这意味着每次操作都会在通话结束后立即返回。

标准Java库中有一个Future接口，但对Netty而言不方便 - 我们只能向Future询问完成操作或阻止当前线程，直到完成操作。这就是为什么Netty有自己的ChannelFuture接口，我们可以将回调传递给ChannelFuture，其将在操作完成时被调用。

Events and Handlers

Netty使用事件驱动的应用程序范例，因此数据处理的管道是通过处理程序的一系列事件。事件和处理程序可以与入站（inbound）和出站（outbound）数据流相关联。 Inbound events（入站事件）可以如下所示：

Channel activation and deactivation
Read operation events
Exception events
User events

Outbound events（出站事件）更简单，通常与打开/关闭连接（ opening/closing a connection）和写入/清空数据（writing/flushing data）有关。

Netty应用程序由几个网络和应用程序逻辑事件及其处理程序组成。通道事件处理程序的基础接口是ChannelHandler及其子接口ChannelOutboundHandler``ChannelInboundHandler。

Netty提供了大量ChannelHandler实现的类。值得注意的是适配器只是空的实现，例如ChannelInboundHandlerAdapter和ChannelOutboundHandlerAdapter。

Encoders and Decoders

当我们使用网络协议时，我们需要执行数据序列化和反序列化。为此，Netty为了能够解码传入数据引入了ChannelInboundHandler的扩展解码器，大多数解码器的基类ByteToMessageDecoder。对于编码输出数据，Netty同样提供了ChannelOutboundHandler的扩展编码器， MessageToByteEncoder是大多数编码器实现的基础。

应用示例

Server Application

创建一个简单协议服务器的项目，它接收请求，执行计算并发送响应。

Dependencies

<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.10.Final</version>
</dependency>
复制代码

Data Model

public class RequestData {private int intValue;private String stringValue;// standard getters and setters
}
复制代码

public class ResponseData {private int intValue;// standard getters and setters
}
复制代码

Request Decoder

It should be noted that Netty works with socket receive buffer, which is represented not as a queue but just as a bunch of bytes. This means that our inbound handler can be called when the full message is not received by a server.

We must make sure that we have received the full message before processing. The decoder for RequestData is shown next:

public class RequestDecoder extends ReplayingDecoder<RequestData> {private final Charset charset = Charset.forName("UTF-8");@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {RequestData data = new RequestData();data.setIntValue(in.readInt());int strLen = in.readInt();data.setStringValue(in.readCharSequence(strLen, charset).toString());out.add(data);}
}
复制代码

ReplayingDecoder It uses an implementation of ByteBuf which throws an exception when there is not enough data in the buffer for the reading operation.

When the exception is caught the buffer is rewound to the beginning and the decoder waits for a new portion of data. Decoding stops when the out list is not empty after decode execution.

Response Encoder

public class ResponseDataEncoder extends MessageToByteEncoder<ResponseData> {@Overrideprotected void encode(ChannelHandlerContext ctx, ResponseData msg, ByteBuf out) throws Exception {out.writeInt(msg.getIntValue());}
}
复制代码

Request Processing

public class ProcessingHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {RequestData requestData = (RequestData) msg;ResponseData responseData = new ResponseData();responseData.setIntValue(requestData.getIntValue() * 2);ChannelFuture future = ctx.writeAndFlush(responseData);future.addListener(ChannelFutureListener.CLOSE);System.out.println(requestData);}
}
复制代码

Server Bootstrap

public class NettyServer {private int port;// constructorpublic static void main(String[] args) throws Exception {int port = args.length > 0? Integer.parseInt(args[0]);: 8080;new NettyServer(port).run();}public void run() throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new RequestDecoder(), new ResponseDataEncoder(), new ProcessingHandler());}}).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);ChannelFuture f = b.bind(port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}
}
复制代码

Client Application

The client should perform reverse encoding and decoding, so we need to have a RequestDataEncoder and ResponseDataDecoder:

public class RequestDataEncoder extends MessageToByteEncoder<RequestData> {private final Charset charset = Charset.forName("UTF-8");@Overrideprotected void encode(ChannelHandlerContext ctx, RequestData msg, ByteBuf out) throws Exception {out.writeInt(msg.getIntValue());out.writeInt(msg.getStringValue().length());out.writeCharSequence(msg.getStringValue(), charset);}
}
复制代码

public class ResponseDataDecoder extends ReplayingDecoder<ResponseData> {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {ResponseData data = new ResponseData();data.setIntValue(in.readInt());out.add(data);}
}
复制代码

Also, we need to define a ClientHandler which will send the request and receive the response from server:

public class ClientHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {RequestData msg = new RequestData();msg.setIntValue(123);msg.setStringValue("all work and no play makes jack a dull boy");ChannelFuture future = ctx.writeAndFlush(msg);}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println((ResponseData)msg);ctx.close();}
}
复制代码

Now let’s bootstrap the client:

public class NettyClient {public static void main(String[] args) throws Exception {String host = "localhost";int port = 8080;EventLoopGroup workerGroup = new NioEventLoopGroup();try {Bootstrap b = new Bootstrap();b.group(workerGroup);b.channel(NioSocketChannel.class);b.option(ChannelOption.SO_KEEPALIVE, true);b.handler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new RequestDataEncoder(), new ResponseDataDecoder(), new ClientHandler());}});ChannelFuture f = b.connect(host, port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();}}
}
复制代码