1. 简介
在本文中，我们将介绍Netty-一个异步事件驱动的网络应用程序框架。
Netty的主要目的是构建基于NIO(或NIO.2)的高性能协议服务器，将网络和业务逻辑组件分离并解耦。它可以实现众所周知的协议，例如HTTP，也可以实现你自己的特定协议。
2. 核心概念
Netty是一个非阻塞框架，与阻塞IO相比，这带来了更高的吞吐量；了解非阻塞IO对于理解Netty的核心组件及其关系至关重要。
2.1 Channel
Channel是Java NIO的基础，它表示一个可以进行读写等IO操作的开放连接。
2.2 Future
Netty中Channel上的每个IO操作都是非阻塞的。
这意味着每个操作在调用后都会立即返回，标准Java库中有一个Future接口，但它对于Netty来说并不方便-我们只能向Future询问操作是否完成，或者阻塞当前线程直到操作完成。
这就是为什么Netty有自己的ChannelFuture接口，我们可以向ChannelFuture传递一个回调，该回调将在操作完成时被调用。
2.3 事件和处理程序
Netty使用事件驱动的应用程序范例，因此数据处理的管道是经过处理程序的一系列事件。事件和处理程序可以与入站和出站数据流相关，入站事件可以是以下几种：
- 通道激活和停用
- 读取操作事件
- 异常事件
- 用户事件
出站事件更简单，通常与打开/关闭连接以及写入/刷新数据有关。
Netty应用程序由几个网络和应用程序逻辑事件及其处理程序组成，通道事件处理程序的基本接口是ChannelHandler及其继承者ChannelOutboundHandler和ChannelInboundHandler。
Netty提供了大量的ChannelHandler实现，值得注意的是适配器只是空实现，例如ChannelInboundHandlerAdapter和ChannelOutboundHandlerAdapter。当我们只需要处理所有事件的子集时，我们可以扩展这些适配器。
此外，还有许多特定协议(如HTTP)的实现，例如HttpRequestDecoder、HttpResponseEncoder、HttpObjectAggregator。
2.4 编码器和解码器
在使用网络协议时，我们需要执行数据序列化和反序列化。为此，Netty引入了ChannelInboundHandler的特殊扩展，用于能够解码传入数据的解码器。大多数解码器的基类是ByteToMessageDecoder。
为了对传出数据进行编码，Netty扩展了ChannelOutboundHandler，称为编码器。MessageToByteEncoder是大多数编码器实现的基础，我们可以使用编码器和解码器将消息从字节序列转换为Java对象，反之亦然。
3. 示例服务器应用程序
让我们创建一个代表简单协议服务器的项目，它接收请求、执行计算并发送响应。
3.1 依赖
首先，我们需要在pom.xml中提供Netty依赖项：
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.10.Final</version>
</dependency>
最新版本可以在Maven Central上找到。
3.2 数据模型
请求数据类具有以下结构：
public class RequestData {
private int intValue;
private String stringValue;
// standard getters and setters
}
假设服务器接收请求并返回乘以2的intValue，响应将具有单个int值：
public class ResponseData {
private int intValue;
// standard getters and setters
}
3.3 请求解码器
现在我们需要为我们的协议消息创建编码器和解码器。
需要注意的是，Netty使用套接字接收缓冲区，它不是以队列的形式表示，而是一堆字节。这意味着当服务器未收到完整消息时，可以调用我们的入站处理程序。
我们必须确保在处理之前已经收到了完整的消息，并且有很多方法可以做到这一点。
首先，我们可以创建一个临时的ByteBuf并将所有入站字节附加到其中，直到获得所需的字节数：
public class SimpleProcessingHandler extends ChannelInboundHandlerAdapter {
private ByteBuf tmp;
@Override
public void handlerAdded(ChannelHandlerContext ctx) {
System.out.println("Handler added");
tmp = ctx.alloc().buffer(4);
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) {
System.out.println("Handler removed");
tmp.release();
tmp = null;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf m = (ByteBuf) msg;
tmp.writeBytes(m);
m.release();
if (tmp.readableBytes() >= 4) {
// request processing
RequestData requestData = new RequestData();
requestData.setIntValue(tmp.readInt());
ResponseData responseData = new ResponseData();
responseData.setIntValue(requestData.getIntValue() * 2);
ChannelFuture future = ctx.writeAndFlush(responseData);
future.addListener(ChannelFutureListener.CLOSE);
}
}
}
上面显示的示例看起来有点奇怪，但有助于我们理解Netty的工作原理。当相应的事件发生时，处理程序的每个方法都会被调用。因此，我们在添加处理程序时初始化缓冲区，在收到新字节时用数据填充它，并在获得足够的数据时开始处理它。
我们故意不使用stringValue-以这种方式解码会不必要地复杂。这就是为什么Netty提供有用的解码器类，它们是ChannelInboundHandler的实现：ByteToMessageDecoder和ReplayingDecoder。
如上所述，我们可以使用Netty创建通道处理管道。因此，我们可以将解码器作为第一个处理程序，处理逻辑处理程序可以位于其后。
RequestData的解码器如下所示：
public class RequestDecoder extends ReplayingDecoder<RequestData> {
private final Charset charset = Charset.forName("UTF-8");
@Override
protected 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);
}
}
这个解码器的思路很简单，它使用ByteBuf的实现，当缓冲区中没有足够的数据进行读取操作时，它会抛出异常。
当捕获到异常时，缓冲区将倒回到开头，解码器将等待新的数据部分。当执行解码后输出列表不为空时，解码将停止。
3.4 响应编码器
除了解码RequestData之外，我们还需要对消息进行编码。此操作更简单，因为在写入操作发生时，我们已经拥有完整的消息数据。
我们可以在主处理程序中将数据写入Channel，或者我们可以分离逻辑并创建一个扩展MessageToByteEncoder的处理程序，该处理程序将捕获写入ResponseData操作：
public class ResponseDataEncoder extends MessageToByteEncoder<ResponseData> {
@Override
protected void encode(ChannelHandlerContext ctx, ResponseData msg, ByteBuf out) throws Exception {
out.writeInt(msg.getIntValue());
}
}
3.5 请求处理
由于我们在单独的处理程序中执行了解码和编码，因此我们需要更改ProcessingHandler：
public class ProcessingHandler extends ChannelInboundHandlerAdapter {
@Override
public 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);
}
}
3.6 服务器启动
现在让我们把它们放在一起并运行我们的服务器：
public class NettyServer {
private int port;
// constructor
public 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>() {
@Override
public 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();
}
}
}
上述服务器启动示例中使用的类的详细信息可以在其Javadoc中找到，最有趣的部分是这一行：
ch.pipeline().addLast(new RequestDecoder(), new ResponseDataEncoder(), new ProcessingHandler());
在这里我们定义入站和出站处理程序，它们将以正确的顺序处理请求和输出。
4. 客户端应用程序
客户端应该执行反向编码和解码，因此我们需要有一个RequestDataEncoder和ResponseDataDecoder：
public class RequestDataEncoder extends MessageToByteEncoder<RequestData> {
private final Charset charset = Charset.forName("UTF-8");
@Override
protected 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> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
ResponseData data = new ResponseData();
data.setIntValue(in.readInt());
out.add(data);
}
}
另外，我们需要定义一个ClientHandler，它将发送请求并接收来自服务器的响应：
public class ClientHandler extends ChannelInboundHandlerAdapter {
@Override
public 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);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println((ResponseData)msg);
ctx.close();
}
}
现在让我们启动客户端：
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>() {
@Override
public 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();
}
}
}
我们可以看到，有很多细节与服务器相同。
现在我们可以运行客户端的main方法并查看控制台输出；正如预期的那样，我们得到了ResponseData，其intValue等于246。
5. 总结
在本文中，我们快速介绍了Netty。我们展示了它的核心组件，例如Channel和ChannelHandler。此外，我们还为其制作了一个简单的非阻塞协议服务器和一个客户端。
Post Directory
