http://www.ox-holdings.com

需要将操作产生的,负责按顺序处理 Channel 中的事件

摘要前段时间,苹果开源了生机勃勃款基于事件驱动的高品质跨平台互联网应用程序开辟框架 SwfitNIO,它微微相似 Netty,但支付语言应用的是 斯维夫特。1、斯威夫特NIO是如何SwfitNIO 实际上是二个底层工具,用于开采高质量的网络应用程序,作为“每连接三个线程”的代表方案。为了提高品质,SwfitNIO 使用了非梗塞 IO,这从它的名字就能够看出来。非堵塞 IO 与梗塞式 IO 特别不少年老成致,因为随意是往互联网上发送数据依旧从网络上接纳数据,应用程序都无须等待,系统内核会在有可操作的 IO 时通报 SwfitNIO。SwfitNIO 并不会提供相近 Web 框架这样的实施方案,而是从事于为上层框架提供底层的营造块。在开拓 Web 应用程序时,一大半开荒者不会一贯运用 SwfitNIO,他们会从 斯维夫特生态系统众多的 Web 框架中甄选多个。不过,那几个框架中的大部分都接纳了 SwfitNIO。2、受支持的阳台SwfitNIO 的对象是扶持具备能够运作 Swift的平台。近期,SwfitNIO 能够在 macOS 和 Linux 上运转,包含:Ubuntu 14.04+macOS 10.12+3、基本架构SwfitNIO 包括了两种基本营造块,全体的 SwfitNIO 应用程序都以由那三种组件组成的。EventLoopGroupEventLoopChannelChannelHandlerBootstrapByteBuffer▶ EventLoopPromise 和 伊夫ntLoopFuture伊夫ntLoop 是 SwfitNIO 最大旨的 IO 原语,它等待事件的产生,在产闯祸变时接触某种回调操作。在超级多 SwfitNIO 应用程序中,EventLoop 对象的数码并十分少,通常每一种 CPU 核数对应意气风发到多个EventLoop 对象。日常的话,伊芙ntLoop 会在应用程序的所有的事生命周期中设有,举行极端的平地风波分发。伊夫ntLoop 可以组合成 EventLoopGroup,伊夫ntLoopGroup 提供了蓬蓬勃勃种机制用来在依次 伊芙ntLoop 间分发专门的工作负荷。比方,服务器在监听外界连接时,用于监听连接的 socket 会被注册到叁个 伊芙ntLoop 上。但大家不希望以此 EventLoop 承受全体的接连几日负载,那么就可以透过 伊芙ntLoopGroup 在四个 伊夫ntLoop 间分摊连接负载。最近,斯维夫特NIO 提供了叁个 伊夫ntLoopGroup 达成(MultiThreaded伊芙ntLoopGroup)和五个 伊夫ntLoop 实现(Selectable伊夫ntLoop 和 Embedded伊夫ntLoop)。MultiThreadedEventLoopGroup 会创设七个线程(使用 POSIX 的 pthreads 库),并为每种线程分配三个 SelectableEventLoop 对象。SelectableEventLoop 使用接收器(基于 kqueue 或 epoll)来治本来自文件和网络 IO 事件。Embedded伊夫ntLoop 是二个空的 EventLoop,什么事也不做,首要用来测量试验。▶ Channels、ChannelHandler、ChannelPipeline 和 ChannelHandlerContext就算伊夫ntLoop 超级重大,但大大多开采者并不会与它有太多的互相,最多正是用它创建伊夫ntLoopPromise 和调整作业。开拓者平常使用的是 Channel 和 ChannelHandler。各种文件汇报符对应一个 Channel,Channel 担任管理文件叙述符的生命周期,并拍卖发生在文书叙述符上的风云:每当 EventLoop 检测到三个与相应的文件叙述符相关的事件,就能布告Channel。ChannelPipeline 由一七种 ChannelHandler 组成,ChannelHandler 担当按顺序管理 Channel 中的事件。ChannelPipeline 就如数据管理管道相通,所以才有了这几个名字。ChannelHandler 要么是 Inbound,要么是 Outbound,要么两个兼有。Inbound 的 ChannelHandler 担负处理“inbound”事件,比如从 socket 读取数据、关闭 socket 大概别的由长途发起的轩然大波。Outbound 的 ChannelHandler 担当管理“outbound”事件,举个例子写多少、发起连接以至关闭本地socket。ChannelHandler 遵照一定顺序处监护人件,例如,读取事件从管道的先头传到前面,而写入事件则从管道的前面传出后边。每一种ChannelHandler 都会在管理完一个平地风波后生成四个新的事件给下三个ChannelHandler。ChannelHandler 是莫斯中国科学技术大学学可接收的零器件,所以尽或者设计得轻量级,各种 ChannelHandler 只处理生机勃勃种多少调换,那样就足以灵活组合各个ChannelHandler,进步代码的可重用性和封装性。我们得以经过 ChannelHandlerContext 来追踪 ChannelHandler 在 ChannelPipeline 中之处。ChannelHandlerContext 富含了近日 ChannelHandler 到上三个和下三个 ChannelHandler 的援用,因而,在任什么时候候,只要 ChannelHandler 还在管道此中,就能够触发新事件。斯威夫特NIO 内置了四种ChannelHandler,包罗 HTTP 拆解深入分析器。别的,SwiftNIO 还提供了有的 Channel 达成,譬如 ServerSocketChannel(用于吸收接纳一而再)、SocketChannel(用于 TCP 连接)、DatagramChannel(用于 UDP socket)和 EmbeddedChannel(用于测量检验)。▶ BootstrapSwiftNIO 提供了部分 Bootstrap 对象,用于简化 Channel 的创始。有些 Bootstrap 对象还提供了别样的局地功效,比如扶植 Happy Eyeballs。近日 斯维夫特NIO 提供了三种 Bootstrap:ServerBootstrap(用于监听 Channel),ClientBootstrap(用于 TCP Channel)和 DatagramBootstrap(用于 UDP Channel)。▶ ByteBufferSwiftNIO 提供了 ByteBuffer,生龙活虎种高效的 Copy-On-Write 字节缓冲器,是绝大大多 SwiftNIO 应用程序的要害构建块。ByteBuffer 提供了多数得力的性状以致一些“钩子”,通过这个钩子,大家能够在“unsafe”的情势下行使 ByteBuffer。这种方法得以博得越来越好的习性,代价是应用程序有十分大希望现身内安抚题。在相符景况下,依然提出在吕梁格局下行使 ByteBuffer。▶ EventLoopPromise 和 伊夫ntLoopFuture并发代码和协助进行代码之间最注重的界别在于无须全体的动作都能够即时到位。譬喻,在向二个Channel 写入数据时,伊芙ntLoop 有异常的大希望不会马上将数据冲刷到互连网上。为此,SwiftNIO 提供了 伊夫ntLoopPromise和 EventLoopFuture,用于管理异步操作。伊芙ntLoopFuture实际上是贰个容器,用于存放函数在以后有些时刻的重回值。种种EventLoopFuture对象都有二个一呼百应的 EventLoopPromise,用于存放实际的结果。只要 EventLoopPromise 推行成功,伊夫ntLoopFuture 也就完了了。通过轮询的艺术检查 EventLoopFuture 是或不是完结是大器晚成种极低效的不二等秘书籍,所以 伊芙ntLoopFuture 被规划成能够选取回调函数。也正是说,在有结果的时候回调函数会被施行。伊夫ntLoopFuture担当管理调解专业,确认保障回调函数是在最早创制伊芙ntLoopPromise 的卓殊 伊夫ntLoop 上进行,所以就从不需求再指向回调函数做其余同步操作。4、斯威夫特NIO 的两全农学SwiftNIO 的指标是要变为强盛的网络应用程序开荒框架,但并不想为全体的档期的顺序抽象提供周密的解决方案。斯威夫特NIO 重要专一在主导的 IO 原语和尾部的情商贯彻上,将别的档期的顺序的架空留给广大的社区去营造。SwiftNIO 将改为服务器端应用程序的构建块,但不料定就是应用程序直接拿来利用的框架。对质量有超级高必要的应用程序恐怕会一贯利用 斯威夫特NIO,收缩上层抽象所拉动的付出。SwiftNIO 能够帮助这么些应用程序在进步品质的还要减少维护资金。斯威夫特NIO 还为有些场景提供了卓有效能的肤浅,高质量的互连网服务器能够直接使用那几个抽象。斯威夫特NIO 的中坚货仓提供了大器晚成部分不行主要的公约落实,比如HTTP。但是,大家以为,超越六分之三协商的贯彻应有要与底层的互连网栈分开,因为它们的发布节奏是很分歧等的。为此,我们鼓舞社区和谐去达成和爱慕他们的磋商贯彻。实际上,SwiftNIO 提供的部分商讨落实开始的一段时期就是由社区支出的,举个例子 TLS 和 HTTP/2。5、相关财富源码托管: 文档: 斯维夫特NIO:聊天客商端: 客户端: 服务器端: 服务器:

客户端

图片 1Figure 9.2 Bootstrap process.jpg

在调用 bind(卡塔尔 或 connect(卡塔尔国 之后,Bootstrap 类肩负创立管道给客商或应用程序,

底层互联网传输 API 必需提须要采取I/O操作的接口,如读,写,连接,绑定等等。对于我们的话,那层布局大约总是会化为一个“socket”。 Netty 中的接口 Channel 定义了与 socket 充足人机联作的操作集:bind, close, config, connect, isActive, isOpen, isWritable, read, write 等等。 Netty 提供大批量的 Channel 实现来特别使用。那一个总结AbstractChannel,AbstractNioByteChannel,AbstractNioChannel,EmbeddedChannel, LocalServerChannel,NioSocketChannel 等等。

ChannelHandler补助广大合计,何况提供用于数据管理的器皿。大家早就通晓ChannelHandler 由特定事件触发。 ChannelHandler 可专用于大概具有的动作,包罗将叁个对象转为字节,试行进度中抛出的不胜管理。常用的一个接口是 ChannelInboundHandler,这一个体系接受到入站事件能够拍卖应用程序逻辑。当您须求提供响合时,你也得以从 ChannelInboundHandler 冲刷数据。一句话,业务逻辑平日存活于二个恐怕多少个ChannelInboundHandler。

ChannelPipeline 提供了四个器皿给 ChannelHandler 链并提供了一个API 用于管理沿着链入站和出站事件的流淌。每种 Channel 都有和好的ChannelPipeline,当 Channel 创造时自动创制的。 ChannelHandler 是什么设置在 ChannelPipeline? 首纵然促成了ChannelHandler 的空洞 ChannelInitializer。ChannelInitializer子类 通过 ServerBootstrap 举办挂号。当它的办法 initChannel(State of Qatar 被调用时,这么些指标将设置自定义的 ChannelHandler 集到 pipeline。当这几个操作完毕时,ChannelInitializer 子类则 从 ChannelPipeline 自动删除自己。

图片 2Figure 3.4 ChannelPipeline with inbound and outbound ChannelHandlers.jpg图片 3Figure 6.3 Channel, ChannelPipeline, ChannelHandler and ChannelHandlerContext.jpg

1.Channel 绑定到 ChannelPipeline2.ChannelPipeline 绑定到 包含 ChannelHandler 的 Channel3.ChannelHandler4.当添加 ChannelHandler 到 ChannelPipeline 时,ChannelHandlerContext 被创建

二、宗旨构件

  • Channel

    Channel是Java NIO的三个主干组织。可以看成是传播或传播数据的载体。因而,它能够被展开或关闭,连接恐怕断开连接。以下是常用的Channel:

-- EmbeddedChannel
-- LocalServerChannel
-- NioDatagramChannel
-- NioSctpChannel
-- NioSocketChannel

  • 回调

    当二个回调被触发时,相应的平地风波能够被一个interface-ChannelHandler的完毕拍卖。

  • Future

    Netty中持有的I/O操作都是异步的。因为二个操作也许不会即时赶回,所以大家需求风流罗曼蒂克种在现在的有些时间点明确其结果的主意。

    Future 和 回调 是相互补充的机制,提供了另生龙活虎种在操作完结时通报应用程序的章程。这一个指标足以看成是一个异步操作结果的占位符;它将在现在的某部时刻做到,并提供对其结果的访谈。

    Netty 提供了ChannelFuture,用于在推行异步操作的时候使用。每一个Netty的出站I/O操作都会回到一个ChannelFuture。ChannelFuture能够注册二个可能八个ChannelFutureListener 实例。监听器的回调方法operationComplete(卡塔尔国,将会在对应的操作实现时被调用。

  • ChannelHandler

    Netty 的第后生可畏组件是ChannelHandler,它充作了具备拍卖入站和出站数据的应用程序逻辑的器皿。

    Netty 使用差别的风浪来公告大家情形的转移仍为操作的景况,每个事件都得以被分发给ChannelHandler类中有些客商实现的方法。Netty提供了大量预订义的可以开箱即用的ChannelHandler完成,包含用于各个协商的ChannelHandler。

    未来,事件能够被分发给ChannelHandler类中某些顾客实现的诀窍。那么,假设ChannelHandler 管理到位后不直接回到给客商端,而是传递给下三个ChannelHandler 继续处理啊?那么将要说起 ChannelPipeline !

    ChannelPipeline 提供了 ChannelHandler链 的器皿,并定义了用来在该链上盛传入站和出站事件流的API。使得事件流经 ChannelPipeline 是 ChannelHandler 的行事,它们是在应用程序的开始化大概教导阶段棉被服装置的。那几个目的摄取事件、执行他们所落成的拍卖逻辑,并将数据传递给链中的下多个ChannelHandler:

1、一个ChannelInitializer的落到实处被登记到了ServerBootstrap中。
2、当 ChannelInitializer.initChannel(卡塔尔方法被调用时, ChannelInitializer就要ChannelPipeline 中设置生机勃勃组自定义的 ChannelHandler。
3、ChannelInitializer 将它和谐从 ChannelPipeline 中移除。

图片 4

  • EventLoop

    伊夫ntLoop 定义了Netty的基本抽象,用来管理连接的生命周期中所产生的轩然大波,在在那之中,将会为各样Channel分配二个伊芙ntLoop。

    EventLoop自己只由三个线程驱动,其拍卖了一个Channel的具备I/O事件,何况在该EventLoop的整个生命周期内都不会转移。那么些轻便而强盛的规划裁撤了您可能有的在ChannelHandler实现中供给进行协同的别的顾忌。

图片 5

 

    这里必要提起,EventLoop的田间管理是因此伊夫ntLoopGroup来实现的。还要一点要专一的是,顾客端指导类是 Bootstrap,只须要三个伊芙ntLoopGroup。服务端指点类是 ServerBootstrap,平日须求五个EventLoopGroup,二个用来接过顾客端连接,三个用来拍卖 I/O 事件(也得以只行使一个 伊夫ntLoopGroup,那时候其将要八个现象下共用同二个伊夫ntLoopGroup)。

图片 6

1、三个 伊芙ntLoopGroup 包涵一个可能几个伊芙ntLoop;
2、一个 伊芙ntLoop 在它的生命周期内只和三个 Thread 绑定;
3、全体由 EventLoop 管理的 I/O 事件都将要它专有的Thread 上被拍卖;
4、多少个 Channel 在它的生命周期内只登记于一个EventLoop;
5、NIO中,三个 伊夫ntLoop 分配给几个Channel(面临八个Channel,三个 伊夫ntLoop 遵照事件触发,顺序试行); OIO中,一个 EventLoop 分配给五个 Channel。

 

  • Bootstrap 和 ServerBootstrap

    BootStarp 和 ServerBootstrap 被喻为引导类,指对应用程序进行配备,并使她运营起来的历程。Netty管理指点的方法是使您的应用程序和互连网层相隔开分离。

    BootStrap 是顾客端的指导类,Bootstrap 在调用 bind(卡塔尔国(连接UDP)和 connect(卡塔尔国(连接TCP)方法时,会新创造一个Channel,仅创设三个单身的、未有父 Channel 的 Channel 来促成全数的互连网沟通。

图片 7

    ServerBootstrap 是服务端的辅导类,ServerBootstarp 在调用 bind()方法时会创造三个 ServerChannel 来选用来自客商端的连接,何况该 ServerChannel 管理了多少个子 Channel 用于同顾客端之间的通讯。

图片 8

 

 tips:Netty 应用程序的多个形似法规:尽只怕的重用 EventLoop,以压缩线程创立所带动的开荒。

EventLoop

  • EventLoop 是用来拍卖 Channel 的 I/O 操作,每多个 Channel 会有一个EventLoop 与它绑定,而且在一切生命周期内都不会退换,而三个 EventLoop 常常会管理三个 Channel,那象征在同二个 伊芙ntLoop 中,四个坦途管理繁忙的话,将会卡住此外通道,因而不要使用阻塞EventLoop,推荐使用 Nio伊芙ntLoop
  • 而多少个 EventLoopGroup 可含蓄两个 EventLoop,不问可以知道伊芙ntLoopGroup 是三个线程池,伊夫ntLoop 是当中三个线程

2、Outbound事件

用来定义Channel能够提供的IO操作。

图片 9

outbound事件列表

ChannelOutboundHandler中定义了Channel提供的进展IO操作的章程。

前两日写了几许netty相关的学识,并写了三个demo,可是对其原理照旧还未深切,前几天我们来做一回商讨吗

一、概念

    开始的一段时期的 Java API 只支持由地面系统套接字库提供所谓的卡住函数来支撑网络编制程序。由于是拥塞 I/O ,要治本四个并发客商端,需求为各样新的顾客端Socket 成立多个 Thread 。那将促成生龙活虎种类的难题,第生机勃勃,在任曾几何时候都可能有大气的线程处于休眠状态(不容许天天都有相应的并发数);第二,需求为各种线程的调用栈都分配内部存款和储蓄器;第三,JVM 在线程的上下文切换所带来的支出会推动劳动。

    Java 在 二〇〇四 年引进了非拥塞 I/O,位于JDK 1.4 的 java.nio 包中。class java.nio.channels.Selector 是Java 的非堵塞 I/O 完成的关键。它采纳了风云通报以分明在黄金时代组非窒碍套接字中有如何已经就绪能够进行I/O 相关的操作。因为能够在其余的小运检查任性的读操作依旧写操作的实现情状,所以如图 1-2 所示,叁个纯净的线程便得以拍卖多少个冒出的连天。

    图片 10

    纵然可以直接行使 Java NIO API,可是在高负荷下可信和赶快地拍卖和调整I/O 操作是大器晚成项繁杂何况轻松失误的职责,最棒还是留给高品质的网络编制程序专家——Netty。

    Netty 是黄金年代款异步的事件驱动的网络应用程序框架,辅助高效的付出可爱抚的高品质的瞄向合同的服务端和客商端。它明白了Java高档API的力量,并将其掩瞒在三个轻便使用的API之后。首先,它的基于 Java NIO 的异步的和事件驱动的贯彻,保障了高负载下应用程序品质的最大化和可伸缩性。其次, Netty 也含有了大器晚成组织设立计情势,将应用程序逻辑从网络层解耦,简化了付出进度, 同一时候也最大限度地加强了可测验性、模块化以至代码的可重用性。

 图片 11

     tips:面向对象的基本概念—> 用较轻松的抽象隐讳底层完成的目不暇接。

Codec

  • 自定义合同其实便是概念解码器和编码器
  • Netty 的 io.netty.handler.codec 包中为大家提供了用来拍卖依赖流(TCP/IP)传输的数量的编码与解码的类,通过它能不负职责字节与 POJO、POJO 与 POJO 的交互作用转换
  • 这么些类精气神上其实就是四个非同一般的 ChannelHandler

1、Inbound事件

用以描述因外表事件以致的Channel状态改造。

图片 12

Inbound事件列表

ChannelInboundHandler中定义了逐生龙活虎Inbound事件的回调方法。

率先让大家来认知一下netty的多少个宗旨人物吧

四、结语

    带着阵阵眼冒Saturn就起头了Netty学习之旅,学现今要么对Netty一群专盛名词头大!不可能,只可以硬着头皮学下来了,终归,熟读唐诗四百首,不会作诗也会吟嘛!

    来总结下,八个Netty服务端管理客商端连接的进度:

1、创制一个channel同该客户端进行绑定;
2、channel从EventLoopGroup得到一个伊芙ntLoop,并登记到该EventLoop,channel生命周期内都和该EventLoop在联合具名(注册时获得selectionKey);
3、channel同顾客端实行互连网连接、关闭和读写,生成相对应的event(改进selectinKey音信),触发eventloop调节线程进行实行;
4、ChannelPipeline 找到呼应 ChannelHandler 方法管理客商逻辑。

    大家项目中动用的 Netty 服务端运行类:

图片 13图片 14

public class NettyServer {

    public static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

    private static Integer LISTENER_PORT = PropertiesLoader.getResourcesLoader().getInteger("nettyPort");



    private int port;
    EventLoopGroup boss = null;
    EventLoopGroup worker = null;
    ServerBootstrap serverBootstrap = null;

    public static NettyServer nettyServer = null;

    public static NettyServer getInstance() {
        if (nettyServer == null) {
            synchronized (NettyServer.class) {
                if (nettyServer == null) {
                    nettyServer = new NettyServer(LISTENER_PORT==null?9999:LISTENER_PORT);
                }
            }
        }
        return nettyServer;
    }

    /**
     * 构造函数
     *
     * @param port 端口
     */
    private NettyServer(int port) {
        this.port = port;

    }

    /**
     * 绑定
     *
     * @throws InterruptedException
     */
    public void init() throws InterruptedException {
        try {

            //创建两个线程池
            //目前服务器CPU为单核8线程,调整线程为8
            boss = new NioEventLoopGroup(8);
            worker = new NioEventLoopGroup(8);

            serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(boss, worker);//两个工作线程
            serverBootstrap.channel(NioServerSocketChannel.class);
            //重用缓冲区
            serverBootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
            serverBootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
            //自动调整下一次缓冲区建立时分配的空间大小,避免内存的浪费
            serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT);
            //当服务器请求处理线程全满时,用于临时存放已完成三次握手的请求的队列的最大长度,默认值50。
            serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            //用于启用或关于Nagle算法。如果要求高实时性,有数据发送时就马上发送,就将该选项设置为true关闭Nagle算法;如果要减少发送次数减少网络交互,就设置为false等累积一定大小后再发送。默认为false。
            serverBootstrap.option(ChannelOption.TCP_NODELAY, true);
            //是否启用心跳保活机制
            serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
            //支持tcp协议
            //bootstrap.childHandler(new TcpChannelInitializer());

            //支持webSocket协议
            serverBootstrap.childHandler(new WebSocketChannelInitializer());
            ChannelFuture f = serverBootstrap.bind(port).sync();
            if (f.isSuccess()) {
                logger.info("netty server start...");
            }
            //等到服务端监听端口关闭
            f.channel().closeFuture().sync();
        } finally {
            //优雅释放线程资源
            boss.shutdownGracefully().sync();
            worker.shutdownGracefully().sync();
        }
    }

    /**
     * 销毁netty相关资源
     */
    public void destroy() {
        try {
            if (boss != null) {
                boss.shutdownGracefully();
            }
            if (worker != null) {
                worker.shutdownGracefully();
            }
            if (serverBootstrap != null) {
                serverBootstrap = null;
            }
        } catch (Exception e) {
            logger.error("netty close err:" + e.getMessage(), e);
        }
    }
}

NettyServer.java

 

参谋资料:《Netty IN ACTION》

示范源代码:https://github.com/JMCuixy/NettyDemo

ChannelInitializer

  • 救助配置客商端连接生成的 Channel,重要责职便是将 ChannelHandler 的贯彻投入到 ChannelPipeline

1、概述

Unsafe是Channel的中间类,三个Channel对应二个Unsafe。

Unsafe用于拍卖Channel对应网络IO的尾部操作。ChannelHandler管理回调事件时产生的相关网络IO操作最后也会委托给Unsafe推行。

服务器端

图片 15Figure 9.3 ServerBootstrap.jpg

ServerChannel实现负勒令立子 Channel,它象征接纳连接

三、实例

    全体的Netty服务端/客商端都最少须求多个部分:

1、最少叁个ChannelHandler —— 该构件完结了对数码的拍卖。

2、指导 —— 那是铺排服务器的开发银行代码。

    服务端:

public class EchoServer {

    private final int port;

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

    public void start() throws InterruptedException {
        final EchoServerHandler serverHandler = new EchoServerHandler();
        //1、创建EventLoopGroup以进行事件的处理,如接受新连接以及读/写数据
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //2、创建ServerBootstrap,引导和绑定服务器
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(group, group)
                    //3、指定所使用的NIO传输Channel
                    .channel(NioServerSocketChannel.class)
                    //4、使用指定的端口设置套接字地址
                    .localAddress(new InetSocketAddress(port))
                    //5、添加一个 EchoServerHandler 到子 Channel的 ChannelPipeline
                    //当一个新的连接被接受时,一个新的子Channel将会被创建,而 ChannelInitializer 将会把一个你的EchoServerHandler 的实例添加到该 Channel 的 ChannelPipeline 中
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(serverHandler);
                        }
                    });
            //6、异步地绑定服务器,调用sync()方法阻塞等待直到绑定完成
            ChannelFuture channelFuture = bootstrap.bind().sync();
            System.out.println(EchoServer.class.getName() + "started and listening for connections on" + channelFuture.channel().localAddress());
            //7、获取 Channel 的 CloseFuture,并且阻塞当前线程直到它完成
            channelFuture.channel().closeFuture().sync();

        } finally {
            //8、关闭 EventLoopGroup 释放所有的资源
            group.shutdownGracefully().sync();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new EchoServer(9999).start();
    }
}

图片 16图片 17

@ChannelHandler.Sharable //标识一个Channel-Handler 可以被多个Channel安全的共享
public class EchoServerHandler extends ChannelHandlerAdapter {


    /**
     * 对于每个传入的消息都要调用
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf in = (ByteBuf) msg;
        System.out.println("Server received:" + in.toString(CharsetUtil.UTF_8));
        //将接收到的消息写给发送者,而不冲刷出站消息
        //ChannelHandlerContext 发送消息。导致消息向下一个ChannelHandler流动
        //Channel 发送消息将会导致消息从 ChannelPipeline的尾端开始流动
        ctx.write(in);
    }

    /**
     * 通知 ChannelHandlerAdapter 最后一次对channel-Read()的调用是当前批量读取中的最后一条消息
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //暂存于ChannelOutboundBuffer中的消息,在下一次调用flush()或者writeAndFlush()方法时将会尝试写出到套接字
        //将这份暂存消息冲刷到远程节点,并且关闭该Channel
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
                .addListener(ChannelFutureListener.CLOSE);
    }

    /**
     * 在读取操作期间,有异常抛出时会调用
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

}

EchoServerHandler.java

 

    客户端:

public class EchoClient {

    private final String host;
    private final int port;

    public EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //创建Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            //指定 EventLoopGroup 以处理客户端事件;适应于NIO的实现
            bootstrap.group(group)
                    //适用于NIO传输的Channel类型
                    .channel(NioSocketChannel.class)
                    .remoteAddress(new InetSocketAddress(host, port))
                    //在创建Channel时,向ChannelPipeline中添加一个EchoClientHandler实例
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new EchoClientHandler());
                        }
                    });
            //连接到远程节点,阻塞等待直到连接完成
            ChannelFuture channelFuture = bootstrap.connect().sync();
            //阻塞,直到Channel 关闭
            channelFuture.channel().closeFuture().sync();
        } finally {
            //关闭线程池并且释放所有的资源
            group.shutdownGracefully().sync();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new EchoClient("127.0.0.1", 9999).start();

        System.out.println("------------------------------------");

        new EchoClient("127.0.0.1", 9999).start();

        System.out.println("------------------------------------");

        new EchoClient("127.0.0.1", 9999).start();
    }


}

图片 18图片 19

@ChannelHandler.Sharable //标记该类的实例可以被多个Channel共享
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    /**
     * 当从服务器接收到一条消息时被调用
     *
     * @param ctx
     * @param msg ByteBuf (Netty 的字节容器) 作为一个面向流的协议,TCP 保证了字节数组将会按照服务器发送它们的顺序接收
     * @throws Exception
     */
    @Override
    protected void messageReceived(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        System.out.println("Client" + ctx.channel().remoteAddress() + "connected");
        System.out.println(msg.toString(CharsetUtil.UTF_8));
    }

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


    /**
     * 在处理过程中引发异常时被调用
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

}

EchoClientHandler.java

ChannelEvent

  • Netty 基于事件驱动,客商的装有 I/O 操作都是三个平地风波,通过对事件的拍卖来完毕大家的业务逻辑

零、目录

风度翩翩、pipeline全体关系简述

二、Unsafe的作用

三、事件的归类及管理

四、pipeline中节点的充足和删除

郑重声明:本文版权归新匍京a奥门-最全网站手机版app官方下载所有,转载文章仅为传播更多信息之目的,如作者信息标记有误,请第一时间联系我们修改或删除,多谢。