Thales的博客站
切换导航
导航
博客首页
关于博主
CV
工具
博客信息
netty 简介
标签
:
java
netty
作者:Thales 尊重博主原创文章,转载请注明文章出于此处。
## 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 Events和Handler Netty使用事件驱动的应用程序范式,因此数据处理的管道是通过处理程序的一系列事件。事件和处理程序可以与入站和出站数据流相关。入站事件可以是以下内容: - Channel激活和停用 - 读取操作事件 - Exception事件 - 用户事件 出站事件更简单,通常与打开/关闭连接和写入/刷新数据有关。 Netty应用程序包含几个网络和应用程序逻辑事件及其handler。Channnel事件处理程序的基接口是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依赖: ```XML
io.netty
netty-all
4.1.10.Final
``` 我们可以在Maven Central上找到最新版本。 ### 3.2 数据模型 请求数据类将具有以下结构: ```java public class RequestData { private int intValue; private String stringValue; // standard getters and setters } ``` 假设服务器接收请求并返回intValue乘以2.Response将只有单个int值: ```java public class ResponseData { private int intValue; // standard getters and setters } ``` ### 3.3 请求解码器 现在我们需要为协议消息创建编码器和解码器。 应该注意的是,Netty使用套接字接收缓冲区,它不是表示队列,而是表示为一堆字节。这意味着当服务器未收到完整消息时,可以调用我们的入站处理程序。 我们必须确保在处理之前收到了完整的消息,并且有很多方法可以做到这一点。 首先,我们可以创建一个临时的ByteBuf并将所有入站字节附加到它,直到我们得到所需的字节数: ```java 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创建一个Channel处理管道。所以我们可以将解码器作为第一个处理程序,处理逻辑处理程序可以在它之后。 RequestData的解码器如下所示: ```java public class RequestDecoder extends ReplayingDecoder
{ private final Charset charset = Charset.forName("UTF-8"); @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List
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的实现,当缓冲区中没有足够的数据用于读取操作时,它会抛出异常。 当捕获到异常时,缓冲区将重绕到开头,解码器将等待新的数据部分。解码执行后,当out列表不为空时,解码停止。 ### 3.4 响应编码器 除解码RequestData外,我们还需要对消息进行编码。此操作更简单,因为在写操作发生时我们有完整的消息数据。 我们可以在主处理程序中将数据写入Channel,或者我们可以分离逻辑并创建一个扩展MessageToByteEncoder的处理程序,它将捕获写入ResponseData操作: ```java public class ResponseDataEncoder extends MessageToByteEncoder
{ @Override protected void encode(ChannelHandlerContext ctx, ResponseData msg, ByteBuf out) throws Exception { out.writeInt(msg.getIntValue()); } } ``` ### 3.5 请求处理 由于我们在单独的处理程序中执行解码和编码,我们需要更改ProcessingHandler: ```java 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 Server BootStrap 现在让我们把它们放在一起并运行我们的服务器: ```java 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
() { @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(); } } } ``` 上面的ServerBootStrap示例中使用的类的详细信息可以在它们的Javadoc中找到。最有趣的部分是这一行: ```java ch.pipeline().addLast( new RequestDecoder(), new ResponseDataEncoder(), new ProcessingHandler()); ``` 在这里,我们定义入站和出站处理程序,它们将以正确的顺序处理请求和输出。 ## 4. 客户端应用 客户端应该执行反向编码和解码,因此我们需要一个RequestDataEncoder和ResponseDataDecoder: ```java public class RequestDataEncoder extends MessageToByteEncoder
{ 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); } } ``` ```java public class ResponseDataDecoder extends ReplayingDecoder
{ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List
out) throws Exception { ResponseData data = new ResponseData(); data.setIntValue(in.readInt()); out.add(data); } } ``` 此外,我们需要定义一个ClientHandler,它将发送请求并从服务器接收响应: ```java 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(); } } ``` 现在让我们bootstrap客户端: ```java 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
() { @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。 ## 总结 在本文中,我们对Netty进行了一个简单的介绍。我们展示它的Channel和ChannelHandler等核心组件。此外,我们做了一个简单的非阻塞协议服务器和客户端。
发布于:『2018-09-04 15:00』
博客类别:
java
阅读(2809) 评论(0)
上一篇:
基于netty的shadowsocks-java(二):原理与结构
下一篇:
使用Lambda快速构建Comparator
用户评论
暂无评论
发表评论
验证码:
发表评论
文章分类
java(10)
Spring(0)
其它(1)
文章存档
2018年10月(1)
2018年09月(8)
2018年07月(2)
友情链接
分享到
暂无评论