Netty简介及使用

一、简介在本文中我们将了解 Netty——一个异步事件驱动的网络应用程序框架。Netty的主要目的是构建基于NIO或者可能是NIO.2的高性能协议服务器网络和业务逻辑组件分离且松散耦合。它可能实现众所周知的协议例如 HTTP或您自己的特定协议。2. 核心概念Netty 是一个非阻塞框架。与阻塞 IO 相比这会带来高吞吐量。理解非阻塞IO对于理解Netty的核心组件及其关系至关重要。2.1.渠道Channel是Java NIO的基础。它代表一个开放的连接能够进行读、写等IO操作。2.2.未来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。如果能在 Netty 的 Javadoc 中熟悉它们就好了。2.4.编码器和解码器当我们使用网络协议时我们需要执行数据序列化和反序列化。为此Netty为能够解码传入数据的解码器引入了ChannelInboundHandler的特殊扩展。大多数解码器的基类是ByteToMessageDecoder。为了对传出数据进行编码Netty 对ChannelOutboundHandler进行了扩展称为编码器。MessageToByteEncoder是大多数编码器实现的基础。我们可以使用编码器和解码器将消息从字节序列转换为 Java 对象反之亦然。3. 示例服务器应用程序让我们创建一个代表简单协议服务器的项目该服务器接收请求、执行计算并发送响应。3.1.依赖关系首先我们需要在pom.xml中提供 Netty 依赖项dependency groupIdio.netty/groupId artifactIdnetty-all/artifactId version4.1.10.Final/version /dependency我们可以在 Maven Central 上找到最新版本。3.2.数据模型请求数据类将具有以下结构public class RequestData { private int intValue; private String stringValue; // standard getters and setters }假设服务器收到请求并返回intValue乘以 2。响应将具有单个 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 ReplayingDecoderRequestData { private final Charset charset Charset.forName(UTF-8); Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, ListObject 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 MessageToByteEncoderResponseData { Override protected void encode(ChannelHandlerContext ctx, ResponseData msg, ByteBuf out) throws Exception { out.writeInt(msg.getIntValue()); } }3.5.请求处理由于我们在单独的处理程序中进行了解码和编码因此我们需要更改我们的ProcessingHandlerpublic 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 ChannelInitializerSocketChannel() { 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和ResponseDataDecoderpublic class RequestDataEncoder extends MessageToByteEncoderRequestData { 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 ReplayingDecoderResponseData { Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, ListObject 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 ChannelInitializerSocketChannel() { 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 方法并查看控制台输出。正如预期的那样我们得到了intValue等于246 的ResponseData。5. 结论在本文中我们快速介绍了 Netty。我们展示了它的核心组件例如Channel和ChannelHandler。此外我们还制作了一个简单的非阻塞协议服务器和一个客户端