暂无图片
暂无图片
暂无图片
暂无图片
暂无图片
首页 / 【Netty】Netty初识篇

【Netty】Netty初识篇

爱编码 2021-07-20
691

后台回复2019赠送1T的学习资料哦！！

简介

官网简介

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

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

不同类型IO和NIO

网络编程当然就少不了IO，所以需要熟悉一下IO基本知识。 下面总结了4种不同类型IO，如下图表。

传统IO和NIO的区别

传统IO如果你想用传统IO来处理多个socket的连接的话需要创建多个线程， 一个线程对应一个，具体反映如下图所示：

NIO而NIO的处理方式是通过多路复用的方式让一个线程去处理多个Socket，这样一来，只需要使用少量的线程就可以搞定多个socket了，线程只需要通过Selector去查一下它所管理的socket集合，哪个Socket的数据准备好了，就去处理哪个Socket，实现了资源最大利用。

NIO特点主要体现： 1.使用较少的线程便可以处理许多连接，因此也减少了内存管理和上下文切换所带来的开销。 2.当没有IO操作需要处理的时候，线程也可以被用于其他任务。

I/O 复用模型

在 I/O 复用模型中，会用到 Select，这个函数也会使进程阻塞，但是和阻塞 I/O 所不同的是这两个函数可以同时阻塞多个 I/O 操作。而且可以同时对多个读操作，多个写操作的 I/O 函数进行检测，直到有数据可读或可写时，才真正调用 I/O 操作函数。

为什么要用Netty

Netty是一个NIO客户端服务器框架，支持快速、简单地开发协议服务器和客户端等网络应用程序。它大大简化和流线网络编程，如TCP和UDP套接字服务器。 Netty经过精心设计，积累了许多协议(如FTP、SMTP、HTTP)的实现经验，以及各种二进制和基于文本的遗留协议。因此，Netty成功地找到了一种方法，在不妥协的情况下实现开发、性能、稳定性和灵活性。

Netty应用场景

暂时总结如下场景：

1.远程过程调用(RPC)，比如dubbo 2.聊天服务器。 3.弹幕服务器。 4.车联网等TCP层网络通讯。 5.Netty与MQTT协议打造LOT服务器。 6.网络游戏服务器。

其实netty和Spring作用非常相似，只是各自的领域不同。如果你是想用Java的TCP层操作选netty准没错。

Netty架构设计

传输服务

支持BIO和NIO

容器集成

支持OSGI、JBossMC、Spring、Guice容器

协议支持

HTTP、Protobuf、二进制、文本、WebSocket等一系列常见协议都支持。 还支持通过实行编码解码逻辑来实现自定义协议

Core核心

可扩展事件模型、通用通信API、支持零拷贝的ByteBuf缓冲对象

Netty之helloworld

这里的代码来自于netty实战，我只是总结一下简单的流程。

客户端

EchoClient.java 是引导和启动客户端

运行流程：

  • 1.为初始化客户端，创建一个Bootstrap实例

  • 2.为进行事件处理分配了一个NioEventLoopGroup实例，其中事件处理包括创建新的连接以及处理入站和出站数据；

  • 3.当连接被建立时，一个EchoClientHandler实例会被安装到（该Channel的一个ChannelPipeline中；

  • 4.在一切都设置完成后，调用Bootstrap.connect()方法连接到远程节

  1. public class EchoClient {

  3.     private final String host;
  4.     private final int port;


  7.     public EchoClient(String host, int port) {
  8.         this.host = host;
  9.         this.port = port;
  10.     }


  13.     /**
  14.      * 运行流程：
  15.      * @param args
  16.      * @throws Exception
  17.      */
  18.     public static void main(String[] args) throws Exception {
  19.         new EchoClient("127.0.0.1",9090).start();
  20.     }

  22.     private void start() throws Exception {

  24.         /**
  25.          * Netty用于接收客户端请求的线程池职责如下。
  26.          * （1）接收客户端TCP连接，初始化Channel参数；
  27.          * （2）将链路状态变更事件通知给ChannelPipeline
  28.          */
  29.         EventLoopGroup group = new NioEventLoopGroup();
  30.         try {
  31.             Bootstrap b = new Bootstrap();
  32.             b.group(group)
  33.                     .channel(NioSocketChannel.class)
  34.                     .remoteAddress(new InetSocketAddress(host,port))
  35.                     .handler(new ChannelInitializer<SocketChannel>() {
  36.                         @Override
  37.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  38.                             socketChannel.pipeline().addLast(new EchoClientHandler());
  39.                         }
  40.                     });
  41.             //绑定端口
  42.             ChannelFuture f = b.connect().sync();

  44.             f.channel().closeFuture().sync();
  45.         } catch (Exception e) {
  46.             group.shutdownGracefully().sync();
  47.         }


  50.     }
  51. }

EchoClientHandler.java 客户端处理类

  1. /**
  2.  * 客户端处理类
  3.  */
  4. public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

  6.     /**
  7.      * 在到服务器的连接已经建立之后将被调用
  8.      * @param ctx
  9.      * @throws Exception
  10.      */
  11.     @Override
  12.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  13.         ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks !", CharsetUtil.UTF_8));
  14.     }

  16.     /**
  17.      * 当从服务器接收到一个消息时被调用
  18.      * @param channelHandlerContext
  19.      * @param byteBuf
  20.      * @throws Exception
  21.      */
  22.     @Override
  23.     protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
  24.         System.out.println("Client received: "+ byteBuf.toString(CharsetUtil.UTF_8));
  25.     }

  27.     /**
  28.      * 在处理过程中引发异常时被调用
  29.      * @param ctx
  30.      * @param cause
  31.      * @throws Exception
  32.      */
  33.     @Override
  34.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  35.         cause.printStackTrace();
  36.         ctx.close();
  37.     }
  38. }

服务端

EchoServer 服务端

运行流程：

  • 1.创建一个ServerBootstrap的实例引导和绑定服务器。

  • 2.创建并分配一个NioEventLoopGroup实例以进行事件的处理，比如接受连接以及读写数据。

  • 3.指定服务器绑定的本地的InetSocketAddress。

  • 4.使用一个EchoServerHandler的实例初始化每一个新的Channel。

  • 5.调用ServerBootstrap.bind()方法以绑定服务器。

  1. public class EchoServer {

  3.     private final int port;

  5.     public EchoServer(int port) {
  6.         this.port = port;
  7.     }


  10.     public static void main(String[] args) throws Exception {
  11.         new EchoServer(9090).start();
  12.     }

  14.     private void start() throws Exception {

  16.         final EchoServerHandler serverHandler = new EchoServerHandler();
  17.         /**
  18.          * NioEventLoop并不是一个纯粹的I/O线程，它除了负责I/O的读写之外
  19.          * 创建了两个NioEventLoopGroup，
  20.          * 它们实际是两个独立的Reactor线程池。
  21.          * 一个用于接收客户端的TCP连接，
  22.          * 另一个用于处理I/O相关的读写操作，或者执行系统Task、定时任务Task等。
  23.          */
  24.         EventLoopGroup bossGroup = new NioEventLoopGroup();
  25.         EventLoopGroup childGroup = new NioEventLoopGroup();

  27.         try {
  28.             //ServerBootstrap负责初始化netty服务器，并且开始监听端口的socket请求
  29.             ServerBootstrap b = new ServerBootstrap();
  30.             b.group(bossGroup, childGroup)
  31.                     .channel(NioServerSocketChannel.class)
  32.                     .localAddress(new InetSocketAddress(port))
  33.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  34.                         @Override
  35.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  36. //                            为监听客户端read/write事件的Channel添加用户自定义的ChannelHandler
  37.                             socketChannel.pipeline().addLast(serverHandler);
  38.                         }
  39.                     });

  41.             ChannelFuture f = b.bind().sync();

  43.             f.channel().closeFuture().sync();
  44.         } catch (InterruptedException e) {
  45.             e.printStackTrace();
  46.         } finally {
  47.             bossGroup.shutdownGracefully().sync();
  48.             childGroup.shutdownGracefully().sync();
  49.         }
  50.     }
  51. }

EchoServerHandler 业务处理handler

  1. /***
  2.  * 服务端自定义业务处理handler
  3.  */
  4. public class EchoServerHandler extends ChannelInboundHandlerAdapter {

  6.     /**
  7.      * 对每一个传入的消息都要调用；
  8.      * @param ctx
  9.      * @param msg
  10.      * @throws Exception
  11.      */
  12.     @Override
  13.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

  15.         ByteBuf in = (ByteBuf) msg;
  16.         System.out.println("server received: "+in.toString(CharsetUtil.UTF_8));

  18.         ctx.write(in);
  19.     }


  22.     /**
  23.      * 通知ChannelInboundHandler最后一次对channelRead()的调用时当前批量读取中的的最后一条消息。
  24.      * @param ctx
  25.      * @throws Exception
  26.      */
  27.     @Override
  28.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  29.         ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
  30.     }

  32.     /**
  33.      * 在读取操作期间，有异常抛出时会调用。
  34.      * @param ctx
  35.      * @param cause
  36.      * @throws Exception
  37.      */
  38.     @Override
  39.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  40.         cause.printStackTrace();
  41.         ctx.close();
  42.     }
  43. }

Netty中部分核心类说明

Bootstrap、ServerBootstrap

Bootstrap意思是引导，一个Netty应用通常由一个Bootstrap开始，主要作用是配置整个Netty程序，串联各个组件，Netty中Bootstrap类是客户端程序的启动引导类，ServerBootstrap是服务端启动引导类。

Future

Future提供了另外一种在操作完成是通知应用程序的方式。这个对象可以看作一个异步操作的结果占位符。通俗地讲，它相当于一位指挥官，发送了一个请求建立完连接，通信完毕了，你通知一声它回来关闭各项IO通道，整个过程，它是不阻塞的，异步的。 在Netty中所有的IO操作都是异步的，不能立刻得知消息是否被正确处理，但是可以过一会等它执行完成或者直接注册一个监听，具体的实现就是通过Future和ChannelFutures，他们可以注册一个监听，当操作执行成功或失败时监听会自动触发注册的监听事件。

Channel

Channel类似Socket，它代表一个实体（如一个硬件设备、一个网络套接字）的开放连接，如读写操作。通俗地讲，Channel字面意思就是通道，每一个客户端与服务端之间进行通信的一个双向通道。

Channel主要工作：

  • 1.当前网络连接的通道的状态（例如是否打开？是否已连接？）

  • 2.网络连接的配置参数 （例如接收缓冲区大小）

  • 3.提供异步的网络 I/O 操作(如建立连接，读写，绑定端口)，异步调用意味着任何 I/O 调用都将立即返回，并且不保证在调用结束时所请求的 I/O 操作已完成。 调用立即返回一个 ChannelFuture 实例，通过注册监听器到ChannelFuture 上，可以 I/O 操作成功、失败或取消时回调通知调用方。

  • 4.支持关联 I/O 操作与对应的处理程序。

不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应，下面是一些常用的 Channel 类型

NioSocketChannel，异步的客户端 TCP Socket 连接 NioServerSocketChannel，异步的服务器端 TCP Socket 连接 NioDatagramChannel，异步的 UDP 连接 NioSctpChannel，异步的客户端 Sctp 连接 NioSctpServerChannel，异步的 Sctp 服务器端连接 这些通道涵盖了 UDP 和 TCP网络 IO以及文件 IO.

EventLoop接口

NioEventLoop中维护了一个线程和任务队列，支持异步提交执行任务，线程启动时会调用NioEventLoop的run方法，执行I/O任务和非I/O任务：I/O任务即selectionKey中ready的事件，如accept、connect、read、write等，由processSelectedKeys方法触发。

非IO任务添加到taskQueue中的任务，如register0、bind0等任务，由runAllTasks方法触发。

两种任务的执行时间比由变量ioRatio控制，默认为50，则表示允许非IO任务执行的时间与IO任务的执行时间相等。

ChannelHanndler

ChannelHandler是一个接口，处理I/O事件或拦截I/O操作，并将其转发到其ChannelPipeline(业务处理链)中的下一个处理程序。 ChannelHandler本身并没有提供很多方法，因为这个接口有许多的方法需要实现，方便使用期间，可以继承它的子类：

ChannelInboundHandler用于处理入站I/O事件 ChannelOutboundHandler用于处理出站I/O操作

或者使用以下适配器类：

ChannelInboundHandlerAdapter用于处理入站I/O事件 ChannelOutboundHandlerAdapter用于处理出站I/O操作 ChannelDuplexHandler用于处理入站和出站事件

参考

《Netty In Action》 《Netty权威指南》

  Netty官网

总结

本文并没有深入学习，仅仅是一些表面的介绍。因为有点忙，所以没空看书，过两天补回。

最后

如果对 Java、大数据感兴趣请长按二维码关注一波，我会努力带给你们价值。觉得对你哪怕有一丁点帮助的请帮忙点个赞或者转发哦。 关注公众号【爱编码】，回复2019有相关资料哦。


数据库
文章转载自爱编码，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

评论