Netty快速入门

Netty是什么

Netty是由JBOSS提供的一个java开源框架，现为 Github上的独立项目。
Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。
Netty主要针对在tcp协议下，面向Clients端的高并发应用，或者Peer-to-Peer场景下大量数据持续传输的应用。
Netty本质是一个NIO框架，适用于服务器通讯相关的多种应用场景。
Netty 可以帮助你快速、简单的开发出一个网络应用，相当于简化和流程化了 NIO 的开发过程
Netty 是目前最流行的 NIO 框架，Netty 在互联网领域、大数据分布式计算领域、游戏行业、通信行业等获得了广泛的应用，知名的 Elasticsearch 、Dubbo 框架内部都采用了 Netty

原生NIO存在的问题

NIO 的类库和 API 繁杂，使用麻烦：需要熟练掌握 Selector、ServerSocketChannel、SocketChannel、ByteBuffer 等。
需要具备其他的额外技能：要熟悉 Java 多线程编程，因为 NIO 编程涉及到 Reactor 模式，必须对多线程和网络编程非常熟悉，才能编写出高质量的 NIO 程序。
开发工作量和难度都非常大：例如客户端面临断连重连、网络闪断、半包读写、失败缓存、网络拥塞和异常流的处理等等。
JDK NIO 的 Bug：例如臭名昭著的 Epoll Bug，它会导致 Selector 空轮询，最终导致 CPU 100%。直到 JDK 1.7 版本该问题仍旧存在，没有被根本解决。

Netty的优点

设计优雅：适用于各种传输类型的统一 API 阻塞和非阻塞 Socket；基于灵活且可扩展的事件模型，可以清晰地分离关注点；高度可定制的线程模型 - 单线程，一个或多个线程池.
使用方便：详细记录的 Javadoc，用户指南和示例；没有其他依赖项，JDK 5（Netty 3.x）或 6（Netty 4.x）就足够了。
高性能、吞吐量更高：延迟更低；减少资源消耗；最小化不必要的内存复制。
安全：完整的 SSL/TLS 和 StartTLS 支持。
社区活跃、不断更新：社区活跃，版本迭代周期短，发现的 Bug 可以被及时修复，同时，更多的新功能会被加入

Netty版本说明

Netty版本分为 netty3.x 和 netty4.x、netty5.x，因为Netty5出现重大bug，已经被官网废弃了，目前推荐使用的是Netty4.x的稳定版本。下载地址：https://bintray.com/netty/downloads/netty/

线程模型基本介绍

目前存在的线程模型有：传统阻塞I/O服务模型、Reactor 模式。根据 Reactor 的数量和处理资源池线程的数量不同，有3种典型的实现：

单 Reactor 单线程；
单 Reactor 多线程；
主从 Reactor 多线程 Netty 线程模式(Netty 主要基于主从 Reactor 多线程模型做了一定的改进，其中主从 Reactor 多线程模型有多个 Reactor)

传统阻塞I/O服务模型

黄色的框表示对象，蓝色的框表示线程，白色的框表示方法(API)。模型特点

采用阻塞IO模式获取输入的数据
每个连接都需要独立的线程完成数据的输入，业务处理，数据返回问题分析
当并发数很大，就会创建大量的线程，占用很大系统资源
连接创建后，如果当前线程暂时没有数据可读，该线程会阻塞在read操作，造成线程资源浪费

Reactor 模式

针对传统阻塞 I/O 服务模型的 2 个缺点，解决方案：

基于 I/O 复用模型：多个连接共用一个阻塞对象，应用程序只需要在一个阻塞对象等待，无需阻塞等待所有连接。当某个连接有新的数据可以处理时，操作系统通知应用程序，线程从阻塞状态返回，开始进行业务处理。Reactor对应的叫法:
1. 反应器模式
2. 分发者模式(Dispatcher)
3. 通知者模式(notifier)
基于线程池复用线程资源：不必再为每个连接创建线程，将连接完成后的业务处理任务分配给线程进行处理，一个线程可以处理多个连接的业务。
Reactor 模式，通过一个或多个输入同时传递给服务处理器的模式(基于事件驱动)
服务器端程序处理传入的多个请求,并将它们同步分派到相应的处理线程，因此Reactor模式也叫Dispatcher模式
Reactor 模式使用IO复用监听事件, 收到事件后，分发给某个线程(进程), 这点就是网络服务器高并发处理关键

Reactor 模式中核心组成

Reactor：Reactor 在一个单独的线程中运行，负责监听和分发事件，分发给适当的处理程序来对 IO 事件做出反应。它就像公司的电话接线员，它接听来自客户的电话并将线路转移到适当的联系人；
Handlers：处理程序执行 I/O 事件要完成的实际事件，类似于客户想要与之交谈的公司中的实际官员。Reactor 通过调度适当的处理程序来响应 I/O 事件，处理程序执行非阻塞操作。

单 Reactor 单线程

理解：单 Reactor 单线程，前台接待员和服务员是同一个人，全程为顾客服

Select 是前面 I/O 复用模型介绍的标准网络编程 API，可以实现应用程序通过一个阻塞对象监听多路连接请求 Reactor 对象通过 Select 监控客户端请求事件，收到事件后通过 Dispatch 进行分发
如果是建立连接请求事件，则由 Acceptor 通过 Accept 处理连接请求，然后创建一个 Handler 对象处理连接完成后的后续业务处理
如果不是建立连接事件，则 Reactor 会分发调用连接对应的 Handler 来响应
Handler 会完成 Read→业务处理→Send 的完整业务流程

优点：模型简单，没有多线程、进程通信、竞争的问题，全部都在一个线程中完成
缺点：性能问题，只有一个线程，无法完全发挥多核 CPU 的性能。Handler 在处理某个连接上的业务时，
整个进程无法处理其他连接事件，很容易导致性能瓶颈
缺点：可靠性问题，线程意外终止，或者进入死循环，会导致整个系统通信模块不可用，
不能接收和处理外部消息，造成节点故障
使用场景：客户端的数量有限，业务处理非常快速，比如Redis在业务处理的时间复杂度O(1)的情况

单Reactor多线程

理解：单 Reactor 多线程，1 个前台接待员，多个服务员，接待员只负责接待

Reactor 对象通过select 监控客户端请求事件, 收到事件后，通过dispatch进行分发
如果建立连接请求, 则由Acceptor 通过accept 处理连接请求, 然后创建一个Handler对象处理完成连接后的各种事件
如果不是连接请求，则由reactor分发调用连接对应的handler 来处理
handler 只负责响应事件，不做具体的业务处理, 通过read 读取数据后，会分发给后面的worker线程池的某个线程处理业务
worker 线程池会分配独立线程完成真正的业务，并将结果返回给handler
handler收到响应后，通过send 将结果返回给client

优点：可以充分的利用多核cpu 的处理能力
缺点：多线程数据共享和访问比较复杂， reactor处理所有的事件的监听和响应，在单线程运行，
在高并发场景容易出现性能瓶颈

主从 Reactor 多线程

理解：主从 Reactor 多线程，多个前台接待员，多个服务生

针对单 Reactor 多线程模型中，Reactor 在单线程中运行，高并发场景下容易成为性能瓶颈，可以让 Reactor 在多线程中运行。

Reactor主线程 MainReactor 对象通过select 监听连接事件, 收到事件后，通过Acceptor 处理连接事件
当 Acceptor 处理连接事件后，MainReactor 将连接分配给SubReactor
subreactor 将连接加入到连接队列进行监听,并创建handler进行各种事件处理
当有新事件发生时， subreactor 就会调用对应的handler处理
handler 通过read 读取数据，分发给后面的worker 线程处理
worker 线程池分配独立的worker 线程进行业务处理，并返回结果
handler 收到响应的结果后，再通过send 将结果返回给client
Reactor 主线程可以对应多个Reactor 子线程, 即MainRecator 可以关联多个SubReactor

优点：父线程与子线程的数据交互简单职责明确，父线程只需要接收新连接，子线程完成后续的业务处理。
优点：父线程与子线程的数据交互简单，Reactor 主线程只需要把新连接传给子线程，子线程无需返回数据。
缺点：编程复杂度较高

Netty模型

Netty抽象出两组线程池， BossGroup 专门负责接收客户端的连接, WorkerGroup 专门负责网络的读写
BossGroup 和 WorkerGroup 类型都是 NioEventLoopGroup
NioEventLoopGroup 相当于一个事件循环组, 这个组中含有多个事件循环，每一个事件循环是 NioEventLoop
NioEventLoop 表示一个不断循环的执行处理任务的线程，每个NioEventLoop 都有一个selector , 用于监听绑定在其上的socket的网络通讯
NioEventLoopGroup 可以有多个线程, 即可以含有多个NioEventLoop
每个Boss NioEventLoop 循环执行的步骤有3步
- 轮询accept事件
- 处理accept事件 , 与client建立连接 , 生成NioScocketChannel , 并将其注册到某个worker NIOEventLoop 上的 selector
- 处理任务队列的任务，即 runAllTasks
每个 Worker NIOEventLoop 循环执行的步骤
- 轮询read, write 事件
- 处理i/o事件，即read , write 事件，在对应NioScocketChannel 处理
- 处理任务队列的任务，即 runAllTasks
每个Worker NIOEventLoop 处理业务时，会使用pipeline(管道), pipeline 中包含了 channel , 即通过pipeline 可以获取到对应通道, 管道中维护了很多的处理器。pipeline就像1个流水线，按顺序执行。

Netty代码示例 - TCP服务

Netty服务端 - Server

java

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class NettyServer {
    public static void main(String[] args) throws Exception{
        /**
         * 创建两个线程组：BossGroup 和 WorkGroup
         * BossGroup(老板)：只负责处理连接请求
         * WorkGroup(小工)：处理客户端业务
         * 这两个线程组都是无限循环
         */
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            //创建服务器端启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();

            //使用链式编程的方式，设置参数
            bootstrap.group(bossGroup,workGroup)    //设置两个线程组
                    .channel(NioServerSocketChannel.class)  //使用NioServerSocketChannel，作为服务器通道实现
                    .option(ChannelOption.SO_BACKLOG,128)   //设置线程队列等待连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE,true)  //设置保持活动连接状态
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        //给pipeline设置处理器
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyServerHandler());
                        }
                    });     //给workGroup的EventLoop对应的管道设置处理器

            //启动服务器，绑定1个端口并且同步，生成了1个ChannelFuture对象
            ChannelFuture cf = bootstrap.bind(6666).sync();

            System.out.println("服务端启动成功~");

            //对关闭通道进行监听
            cf.channel().closeFuture().sync();
        } finally {
            //优雅关闭
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

Netty服务端 - Handler

java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;

//自定义的handler，需要继承netty 规定好的某个HandlerAdapter
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 读取数据
     * ChannelHandlerContext:上下文对象，含有pipeline和channel通道、地址等
     * msg：客户端发送的数据，默认Object
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //将msg消息，转成ByteBuf，ByteBuf是netty提供的
        ByteBuf buf = (ByteBuf)msg;
        System.out.println("客户端发来消息：" + buf.toString(CharsetUtil.UTF_8));
        //注意：使用完ByteBuf之后，需要主动去释放资源，否则，资源一直在内存中加载，容易造成内存泄漏
        ReferenceCountUtil.release(buf);
    }

    /**
     * 数据读取完毕
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //将数据写入到缓冲并刷新，write + flush
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello，我是服务端",CharsetUtil.UTF_8));
    }

    /**
     * 发生异常处理，一般关闭通道
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

}

Netty客户端 - Client

java

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyClient {
    public static void main(String[] args) throws Exception{
        //客户端只需要一个时间循环组
        NioEventLoopGroup group = new NioEventLoopGroup();

        try {
            //创建客户端启动对象，客户端使用Bootstrap
            Bootstrap bootstrap = new Bootstrap();

            //设置相关参数
            bootstrap.group(group)  //设置线程组
                    .channel(NioSocketChannel.class)    //设置客户端通道实现
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyClientHandler());
                        }
                    });

            //启动客户端去连接服务端
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666).sync();

            //给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } finally {
            //优雅关闭
            group.shutdownGracefully();
        }

    }
}

Netty客户端 - Handler

java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;

//自定义的handler，需要继承netty 规定好的某个HandlerAdapter
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    //当通道就绪，就会触发该方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("通道就绪");
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello，我是客户端", CharsetUtil.UTF_8));
    }

    //当通道有读取时间时触发
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //将msg消息，转成ByteBuf，ByteBuf是netty提供的
        ByteBuf buf = (ByteBuf)msg;
        System.out.println("服务端发来消息：" + buf.toString(CharsetUtil.UTF_8));
        //注意：使用完ByteBuf之后，需要主动去释放资源，否则，资源一直在内存中加载，容易造成内存泄漏
        ReferenceCountUtil.release(buf);
    }

    //异常时触发
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

Netty代码示例 - HTTP服务

HttpServer

java

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpServerCodec;

public class HttpServer {
    public static void main(String[] args) throws Exception{
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup,workGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //得到管道
                            ChannelPipeline pipeline = ch.pipeline();
                            //加入一个netty提供的 HttpServerCodec 编解码
                            pipeline.addLast("MyHttpServerCodec",new HttpServerCodec());
                            //增加 1 个自定义 handler
                            pipeline.addLast("MyServerHandler",new HttpServerHandler());
                        }
                    });

            ChannelFuture channelFuture = bootstrap.bind(6666).sync();

            System.out.println("服务端启动完毕！");

            channelFuture.channel().closeFuture().sync();
        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

HttpServerHandler

java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;
import java.net.URI;

/**
 * SimpleChannelInboundHandler 它是ChannelInboundHandlerAdapter子类
 * HttpObject 表示客户端和服务器端相互通讯的数据被封装成 HttpObject
 */
public class HttpServerHandler extends SimpleChannelInboundHandler<HttpObject> {

    //channelRead0 读取客户端数据
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
        //判断 msg 是不是一个 HttpRequest 请求
        if(msg instanceof HttpRequest){
            System.out.println("客户端地址：" + ctx.channel().remoteAddress());

            //获取URL，对特定资源进行过滤
            HttpRequest request = (HttpRequest) msg;
            URI uri = new URI(request.uri());
            if("/favicon.ico".equals(uri.getPath())){
                System.out.println("请求了 favicon.ico(图标) 资源，不做处理");
                return;
            }

            //回复信息浏览器
            ByteBuf content = Unpooled.copiedBuffer("Hello，我是服务端！", CharsetUtil.UTF_8);
            //构造一个 HttpResponse
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
            //设置返回类型
            response.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/plain");
            //设置返回长度
            response.headers().set(HttpHeaderNames.CONTENT_LENGTH,content.readableBytes());
            //将构建好的response返回
            ctx.writeAndFlush(response);
            //这里继承 SimpleChannelInboundHandler 类，则不需要释放ByteBuf，因为该类已经帮着释放了
        }
    }
}

代码测试

浏览器访问：127.0.0.1:6666

异步模型

异步的概念和同步相对。当一个异步过程调用发出后，调用者不能立刻得到结果。实际处理这个调用的组件在完成后，通过状态、通知和回调来通知调用者。
Netty 中的 I/O 操作是异步的，包括 Bind、Write、Connect 等操作会简单的返回一个 ChannelFuture。
调用者并不能立刻获得结果，而是通过 Future-Listener 机制，用户可以方便的主动获取或者通过通知机制获得 IO 操作结果
Netty 的异步模型是建立在 future 和 callback 的之上的。callback 就是回调。重点是 Future，它的核心思想是：假设一个方法 fun，计算过程可能非常耗时，等待 fun返回显然不合适。那么可以在调用 fun 的时候，立马返回一个 Future，后续可以通过 Future去监控方法 fun 的处理过程(即： Future-Listener 机制)

taskQueue自定义任务

java

//用户程序自定义的普通任务
//Handler中的read方法
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //比如这里有一个非常耗费时间的业务，耗时业务
    ctx.channel().eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(10 * 1000);//模拟耗时业务 10秒
            }catch (Exception e){
                e.printStackTrace();
            }
            ctx.writeAndFlush(Unpooled.copiedBuffer("Hello，我是服务端",CharsetUtil.UTF_8));
        }
    });
    System.out.println("此处直接输出，不会阻塞。");
}

scheduleTaskQueue自定义定时任务

java

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //比如这里有一个非常耗费时间的业务，耗时业务

    //var1：执行线程 var2：初始化延时 var4：两次开始执行最小间隔时间  var6：计时单位
    ctx.channel().eventLoop().scheduleAtFixedRate(Runnable var1, long var2, long var4, TimeUnit var6)
  
    //var1：执行线程 var2：初始化延时 var4：前一次执行结束到下一次执行开始的间隔时间（间隔执行延迟时间） var6：计时单位
    ctx.channel().eventLoop().scheduleWithFixedDelay(Runnable var1, long var2, long var4, TimeUnit var6)
  
    //command：执行线程 delay：初始化延时 unit：计时单位
    ctx.channel().eventLoop().schedule(Runnable command,long delay, TimeUnit unit)
  
    //callable：执行线程 delay：初始化延时 unit：计时单位
    ctx.channel().eventLoop().schedule(Callable<V> callable,long delay, TimeUnit unit)

}

Bootstrap、ServerBootstrap

Bootstrap 意思是引导，一个 Netty 应用通常由一个 Bootstrap 开始，主要作用是配置整个 Netty 程序，串联各个组件，**Netty 中 Bootstrap 类是客户端程序的启动引导类，ServerBootstrap 是服务端启动引导类 **
常用的方法：
- public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup)，该方法用于服务器端，用来设置两个 EventLoop
- public B group(EventLoopGroup group) ，该方法用于客户端，用来设置一个 EventLoop
- public B channel(Class<? extends C> channelClass)，该方法用来设置一个服务器端的通道实现
- public <T> B option(ChannelOption<T> option, T value)，用来给 ServerChannel 添加配置
- public <T> ServerBootstrap childOption(ChannelOption<T> childOption, T value)，用来给接收到的通道添加配置
- public ServerBootstrap childHandler(ChannelHandler childHandler)，该方法用来设置业务处理类（自定义的 handler）
- public ChannelFuture bind(int inetPort) ，该方法用于服务器端，用来设置占用的端口号
- public ChannelFuture connect(String inetHost, int inetPort) ，该方法用于客户端，用来连接服务器端

Future、ChannelFuture

Netty 中的Future是一个接口，它继承了JUC下的Future。它的addListener方法比较重要，它的sync方法会阻塞直到future完成操作，如果操作失败会重新抛出异常。future中有get方法来获取异步的结果，get方法会阻塞后续的逻辑代码，如果使用addListener则不会。Netty中的ChannelFuture继承了Netty中的自己的Future， public interface ChannelFuture extends Future<Void>，ChannelFuture表示Channel中异步I/O操作的结果，在netty中所有的I/O操作都是异步的，I/O的调用会直接返回，可以通过ChannelFuture来获取I/O操作的结果状态。常用方法：

Channel channel()，返回当前正在进行 IO 操作的通道
ChannelFuture sync()，等待异步操作执行完毕，调用的sync()的目的就是保证ChannelFuture已经完成
addListener()，addListener方法传入的监听器会实现GenericFutureListener接口，也就是被通知的时候operationComplete方法会被调用

在使用 Netty 进行编程时，拦截操作和转换出入站数据只需要提供 callback 或利用future 即可。这使得链式操作简单、高效, 并有利于编写可重用的、通用的代码。
Netty 框架的目标就是让你的业务逻辑从网络基础应用编码中分离出来、解脱出来

Future-Listener 机制

当 Future 对象刚刚创建时，处于非完成状态，调用者可以通过返回的 ChannelFuture 来获取操作执行的状态，注册监听函数来执行完成后的操作。
常见有如下操作
- 通过 isDone 方法来判断当前操作是否完成；
- 通过 isSuccess 方法来判断已完成的当前操作是否成功；
- 通过 getCause 方法来获取已完成的当前操作失败的原因；
- 通过 isCancelled 方法来判断已完成的当前操作是否被取消；
- 通过 addListener 方法来注册监听器，当操作已完成(isDone 方法返回完成)，将会通知指定的监听器；如果 Future 对象已完成，则通知指定的监听器
代码示例

java

bootstrap.bind(6666).addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if(cf.isSuccess()){
            System.out.println("监听端口6666成功！");
        }
    }
});

相比传统阻塞 I/O，执行 I/O 操作后线程会被阻塞住, 直到操作完成；
异步处理的好处是不会造成线程阻塞，线程在 I/O 操作期间可以执行别的程序，
在高并发情形下会更稳定和更高的吞吐量

Channel

Channel是Netty 网络通信的组件，能够用于执行网络 I/O 操作。
通过Channel 可获得当前网络连接的通道的状态
通过Channel 可获得网络连接的配置参数（例如接收缓冲区大小）
Channel 提供异步的网络 I/O 操作(如建立连接，读写，绑定端口)，异步调用意味着任何 I/O 调用都将立即返回，并且不保证在调用结束时所请求的 I/O 操作已完成
调用立即返回一个 ChannelFuture 实例，通过注册监听器到 ChannelFuture 上，可以 I/O 操作成功、失败或取消时回调通知调用方
支持关联 I/O 操作与对应的处理程序
不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应，常用的 Channel 类型:
- NioSocketChannel，异步的客户端 TCP Socket 连接。
- NioServerSocketChannel，异步的服务器端 TCP Socket 连接。
- NioDatagramChannel，异步的 UDP 连接。
- NioSctpChannel，异步的客户端 Sctp 连接。
- NioSctpServerChannel，异步的 Sctp 服务器端连接，这些通道涵盖了 UDP 和 TCP 网络 IO 以及文件 IO。

ChannelHandler

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

java

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {
    public ChannelInboundHandlerAdapter() { }
    //channel注册事件
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelRegistered();
    }
    //channel取消注册事件
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelUnregistered();
    }
    //通道就绪事件-请求进入时触发  
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();
    }
    //连接关闭时触发
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelInactive();
    }
    //通道读取数据事件
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.fireChannelRead(msg);
    }
    //数据读取完毕事件
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelReadComplete();
    }
    //用户事件触发(读或者写，心跳监测)
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        ctx.fireUserEventTriggered(evt);
    }
    //在 channel 写状态发生变化时调用
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelWritabilityChanged();
    }
    //通道发生异常事件
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.fireExceptionCaught(cause);
    }
}

出站/入站

java

p.addLast("1", new InboundHandlerA());
p.addLast("2", new InboundHandlerB());
p.addLast("3", new OutboundHandlerA());
p.addLast("4", new OutboundHandlerB());
p.addLast("5", new InboundOutboundHandlerX());

ChannelInboundHandler(处理输入字节)
ChannelOutboundHandler（处理输出字节）

从代码中看出ChannelHandler在ChannelPipeline中的编排顺序是从1，2，3，4，5
当入站时是从ChannelPipeline中从头部依次执行的，也就是依次执行1，2，5
当出站时是从ChannelPipeline中从尾部向头部的方向依次执行，也就是依次执行5，4，3
例如：
从一个客户端应用程序的角度看，如果事件的运动方向是客户端到服务端，
那么称这些事件为出站的，反之则称为入站。
同理，从一个服务端的角度看，如果事件的运动方向是服务端到客户端则称为出站，反之为入站。

Pipeline 和 ChannelPipeline

ChannelPipeline 是一个 Handler 的集合，它负责处理和拦截 inbound 或者 outbound 的事件和操作，相当于一个贯穿 Netty 的链。(也可以这样理解：ChannelPipeline 是保存 ChannelHandler 的 List，用于处理或拦截 Channel 的入站事件和出站操作)
ChannelPipeline 实现了一种高级形式的拦截过滤器模式，使用户可以完全控制事件的处理方式，以及 Channel 中各个的 ChannelHandler 如何相互交互
在 Netty 中每个 Channel 都有且仅有一个 ChannelPipeline 与之对应，它们的组成关系如下

一个 Channel 包含了一个 ChannelPipeline，而 ChannelPipeline 中又维护了一个由 ChannelHandlerContext 组成的双向链表，并且每个 ChannelHandlerContext 中又关联着一个 ChannelHandler
入站事件和出站事件在一个双向链表中，入站事件会从链表 head 往后传递到最后一个入站的 handler，出站事件会从链表 tail 往前传递到最前一个出站的 handler，两种类型的 handler 互不干扰
常用方法
- ChannelPipeline addFirst(ChannelHandler... handlers)，把一个业务处理类（handler）添加到链中的第一个位置
- ChannelPipeline addLast(ChannelHandler... handlers)，把一个业务处理类（handler）添加到链中的最后一个位置

ChannelHandlerContext ctx

保存 Channel 相关的所有上下文信息，同时关联一个 ChannelHandler 对象
即ChannelHandlerContext 中包含一个具体的事件处理器 ChannelHandler ，同时ChannelHandlerContext 中也绑定了对应的 pipeline 和 Channel 的信息，方便对 ChannelHandler进行调用
常用方法
- ChannelFuture close()，关闭通道
- ChannelOutboundInvoker flush()，刷新
- ChannelFuture writeAndFlush(Object msg) ，将数据写到 ChannelPipeline 中当前 ChannelHandler的下一个 ChannelHandler 开始处理（出站）

option和childOption

区别option和childOption，简单就是说，在服务端程序中，option是针对自己的服务配置（监听socket使用），childOption是客户端连接成功后的channel配置选项（客户端连接后使用）。ChannelOption常用参数:

java

bootstrap.option(ChannelOption.SO_BACKLOG,4096);//请求的队列的最大长度
bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);//重用缓冲区-服务端
bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);//重用缓冲区-客户端
bootstrap.childOption(ChannelOption.TCP_NODELAY, true);//立即发送数据
bootstrap.option(ChannelOption.SO_RCVBUF,128);//接收缓冲区大小
bootstrap.option(ChannelOption.SO_SNDBUF,256);//发送缓冲区大小
bootstrap.option(ChannelOption.SO_REUSEADDR,true);//快速端口复用
bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);//连接保活

EventLoopGroup 和其实现类 NioEventLoopGroup

EventLoopGroup 是一组 EventLoop 的抽象，Netty 为了更好的利用多核 CPU 资源，一般会有多个 EventLoop 同时工作，每个 EventLoop 维护着一个 Selector 实例。
EventLoopGroup 提供 next 接口，可以从组里面按照一定规则获取其中一个 EventLoop来处理任务。在 Netty 服务器端编程中，我们一般都需要提供两个 EventLoopGroup，例如：BossEventLoopGroup 和 WorkerEventLoopGroup。
通常一个服务端口即一个 ServerSocketChannel对应一个Selector 和一个EventLoop线程。BossEventLoop 负责接收客户端的连接并将 SocketChannel 交给 WorkerEventLoopGroup 来进行 IO 处理

1.BossEventLoopGroup 通常是一个单线程的 EventLoop，EventLoop 维护着一个注册了
ServerSocketChannel 的 Selector 实例BossEventLoop 不断轮询 Selector 将连接事件分离出来
2.通常是 OP_ACCEPT 事件，然后将接收到的 SocketChannel 交给 WorkerEventLoopGroup
3.WorkerEventLoopGroup 会由 next 选择其中一个 EventLoop 来将这个 SocketChannel 
注册到其维护的 Selector 并对其后续的 IO 事件进行处理

常用方法
public NioEventLoopGroup()，构造方法
public Future<?> shutdownGracefully()，断开连接，关闭线程

Unpooled 类

Netty 提供一个专门用来操作缓冲区（即Netty的数据容器ByteBuf）的工具类，常用方法：

java

//通过给定的数据和字符编码返回一个 ByteBuf 对象（类似于 NIO 中的 ByteBuffer 但有区别）
public static ByteBuf copiedBuffer(CharSequence string, Charset charset)


使用说明：
示例1：
/**
 * 1.创建1个对象，该对象包含1个数组arr，是一个byte[10]
 * 2.netty的ByteBuf的读写不需要flip()反转，因为底层维护了1个 readerIndex 和 writerIndex 和 capacity
 * readerIndex 表示下一个读取的位置
 * writerIndex 表示下一个写入的位置
 * capacity 表示长度
 */
ByteBuf buffer = Unpooled.buffer(10);
for (int i = 0; i < 10; i++) {
    buffer.writeByte(i);
}
System.out.println("buffer长度：" + buffer.capacity());
//输出
for (int i = 0; i < buffer.capacity(); i++) {
    //buffer.readByte() - 顺序读取
    System.out.println("值：" + buffer.readByte());
    //buffer.getByte(i) - 指定下标读取
    //System.out.println("值：" + buffer.getByte(i));
}


示例2：
ByteBuf buffer = Unpooled.copiedBuffer("Hello,世界！",CharsetUtil.UTF_8);
//使用相关API方法
if(buffer.hasArray()){//buffer.hasArray() - 缓冲区中有数组时返回true
    //得到数组
    byte[] content = buffer.array();
    //将content转成字符串
    System.out.println(new String(content, Charset.forName("utf-8")));
    System.out.println(buffer.arrayOffset());//数组偏移量
    System.out.println(buffer.readerIndex());//下一个读取位置索引
    System.out.println(buffer.writerIndex());//下一个写入位置索引
    System.out.println(buffer.capacity());//总长度
    buffer.readByte();//读取1个字节，该操作会影响readerIndex
    buffer.getByte(0);//指定索引位置读取，该操作读取不会影响readerIndex
    int count = buffer.readableBytes();//可读字节数
    //CharSequence：String 继承于CharSequence，CharSequence 是 char 值的一个可读序列
    CharSequence charSequence = buffer.getCharSequence(0, 5, CharsetUtil.UTF_8);//指定起始和结束索引读取，该操作不会影响readerIndex
}

Netty代码示例 - 消息群聊

群聊服务端 - Server

java

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class GroupCharServer {

    private int port;   //监听端口
    public GroupCharServer(int port){
        this.port = port;
    }
    //处理客户端请求
    public void run() throws Exception{
        //创建2个线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup(10);
        try {
            //创建bootstrap
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,1024)
                    .childOption(ChannelOption.SO_KEEPALIVE,true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //获取到pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            //解码器
                            pipeline.addLast("decoder",new StringDecoder());
                            //业务处理handler
                            pipeline.addLast(null);
                            //编码器
                            pipeline.addLast("encoder",new StringEncoder());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.bind(port).sync();
            System.out.println("Netty服务端启动！");
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception{
        new GroupCharServer(6666).run();
    }
}

群聊服务端 - handler

java

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

public class GroupCharServerHandler extends SimpleChannelInboundHandler<String> {

    //定义 1 个channel组，管理所有的channel
    //GlobalEventExecutor.INSTANCE 是全局的事件执行器，是一个单例
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    //消息接收后处理
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        Channel channel = ctx.channel();
        //根据不同的情况，返回不同的消息
        channelGroup.forEach(ch -> {
            if(ch != channel){
                //如果不是当前的channel（自己），直接转发
                ch.writeAndFlush("[游客]" + channel.remoteAddress() + " -> 发来消息：" + msg + "\n");
            } else {
                //如果是自己，消息回显
                ch.writeAndFlush("[自己] -> 发送消息：" + msg + "\n");
            }
        });
    }

    //异常时触发，关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

    //handlerAdded 表示连接建立，一旦连接建立，第一个被执行
    //连接建立，将当前的channel加入到channelGroup
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        //加入之前，将该客户端加入聊天的信息，推送给其他在线的客户端
        //channelGroup.writeAndFlush 该方法会将channelGroup中所有的channel遍历，并发送消息
        channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " -> 加入聊天\n");
        channelGroup.add(channel);
    }

    //handlerRemoved 表示断开连接触发
    //断开连接，删掉该channel
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " -> 离开聊天\n");
        //触发该方法，channelGroup会自动删掉该channel
    }

    //channelActive 表示channel处于活动状态
    //提示 XXX 上线
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + " -> 上线了");
    }

    //channelInactive 表示channel处于非活动状态触发
    //提示 XXX 离线
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + " -> 下线了");
    }
}

群聊客户端 - Client

java

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import java.util.Scanner;

public class GroupCharClient {

    private final String host;
    private final int port;
    public GroupCharClient(String host,int port){
        this.host = host;
        this.port = port;
    }
    public void run() throws Exception{
        NioEventLoopGroup eventExecutors = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventExecutors)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //解码器
                            pipeline.addLast("decoder",new StringDecoder());
                            //业务处理handler
                            pipeline.addLast(new GroupCharClientHandler());
                            //编码器
                            pipeline.addLast("encoder",new StringEncoder());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.connect(host,port).sync();
            System.out.println("客户端连接成功！");
            //得到channel
            Channel channel = channelFuture.channel();
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()){
                String msg = scanner.nextLine();
                //通过channel发送到服务端
                channel.writeAndFlush(msg + "\n");
            }
        }finally {
            eventExecutors.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception{
        new GroupCharClient("127.0.0.1",6666).run();
    }

}

群聊客户端 - handler

java

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class GroupCharClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(msg.trim());
    }
}

心跳监测

java

//在服务端的pipeline中加入 IdleStateHandler
//参数1：readerIdleTime-表示channel多久没有读操作，就会发送1个心跳检测包，检测是否还是连接状态
//参数2：writerIdleTime-表示channel多久没有写操作，就会发送1个心跳检测包，检测是否还是连接状态
//参数3：allIdleTime-表示channel多久没有读和写操作，就会发送1个心跳检测包，检测是否还是连接状态
//参数4：时间单位
pipeline.addLast(new IdleStateHandler(3,5,7, TimeUnit.SECONDS));

当IdleStateHandler触发后，就会传递给管道下一个handler去处理，
通过调用(触发)下一个handler的 userEventTiggered() 方法，
所以在它的下面，加入处理读、写或读和写空闲的处理handler
pipeline.addLast(new TimeoutHandler());

TimeoutHandler

java

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    //判断是否属于 IdleStateEvent
    if(evt instanceof IdleStateEvent){
        IdleStateEvent event = (IdleStateEvent) evt;
        //判断事件
        switch (event.state()) {
            case READER_IDLE:
                System.out.println("读空闲处理");
                break;
            case WRITER_IDLE:
                System.out.println("写空闲处理");
                break;
            case ALL_IDLE:
                System.out.println("读写空闲处理");
                break;
        }
    }
}

Netty代码示例 - WebSocket长连接

WebSocket长连接服务端 - Server

java

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WebsocketServer {
    public static void main(String[] args) throws Exception{
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup,workGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //得到管道
                            ChannelPipeline pipeline = ch.pipeline();

                            //加入一个netty提供的 HttpServerCodec 编解码
                            pipeline.addLast(new HttpServerCodec());

                            //以块方式写，添加ChunkedWrite处理器
                            pipeline.addLast(new ChunkedWriteHandler());

                            //HttpObjectAggregator 可以将多个段聚合起来，比如浏览器发送大量数据时，就会发出多次http请求
                            pipeline.addLast(new HttpObjectAggregator(8000));

                            /**
                             * 1.对于websocket，它的数据是以 帧(frame) 形式传递
                             * 2.浏览器请求时 ws://localhost:6666/web  (/web 表示请求url，可自定义)
                             * 3.WebSocketServerProtocolHandler 核心功能将http协议升级为WebSocket协议，保持长连接
                             * 4.是通过状态码来升级的。状态码 101 升级为WebSocket
                             */
                            pipeline.addLast(new WebSocketServerProtocolHandler("/web"));

                            //自定义 handler - 业务逻辑处理
                            pipeline.addLast(new WebsocketServerHandler());
                        }
                    });

            ChannelFuture channelFuture = bootstrap.bind(6666).sync();

            System.out.println("服务端启动完毕！");

            channelFuture.channel().closeFuture().sync();
        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

WebSocket长连接 - handler

java

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import java.time.LocalDateTime;

//TextWebSocketFrame 表示一个文本帧(Frame)
public class WebsocketServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        System.out.println("服务端收到消息" + msg.text());
        //回复消息
        ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间：" + LocalDateTime.now() + " -> " + msg.text()));
    }

    //当web客户端连接后，触发该方法
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        //id()表示唯一值，它有两种形式，LongText 长值    ShortText 短值
        System.out.println("handlerAdded被调用：" + ctx.channel().id().asLongText());
        System.out.println("handlerAdded被调用：" + ctx.channel().id().asShortText());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println("handlerRemoved被调用：" + ctx.channel().id().asLongText());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("发生异常：" + cause.getMessage());
        ctx.close();
    }
}

WebSocket长连接 - 页面调用

html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>WebSocket</title>
</head>
<body>
<script>
    var websocket;
    if(window.WebSocket){
        websocket = new WebSocket("ws://localhost:6666/web");
        //相当于channelRead0 ， 收到服务器端回送的消息
        websocket.onmessage = function (msg) {
            var response = document.getElementById('responseText');
            response.value = response.value + "\n" + msg.data;//回显
        }
        //相当于连接开启
        websocket.onopen = function (conn) {
            var response = document.getElementById('responseText');
            response.value = '连接已开启......';
        }
        //相当于连接关闭
        websocket.onclose = function (conn) {
            var response = document.getElementById('responseText');
            response.value = response.value + "\n" + '连接已关闭......';
        }
    } else {
        alert("当前浏览器不支持WebSocket");
    }

    //发送消息给服务端
    function send(msg) {
        if(!window.websocket){
            //如果未创建websocket，直接返回
            return;
        }
        //如果是开启状态
        if(websocket.readyState == WebSocket.OPEN){
            //发送消息
            websocket.send(msg);
        } else {
            alert("连接未开启！")
        }
    }
</script>
<form onsubmit="return false">
    <textarea name="msg" rows="20" cols="100"></textarea>
    <input type="button" value="发送" onclick="send(this.form.msg.value)" />

    <textarea id="responseText" rows="20" cols="100"></textarea>
    <input type="button" value="清空" onclick="document.getElementById('responseText').value=''" />
</form>
</body>
</html>

编码和解码

编写网络应用程序时，因为数据在网络中传输的都是二进制字节码数据，在发送数据时就需要编码，接收数据时就需要解码。codec(编解码器) 的组成部分有两个：decoder(解码器)和 encoder(编码器)。encoder 负责把业务数据转换成字节码数据，decoder 负责把字节码数据转换成业务数据

Netty 本身的编码解码的机制和问题

无法跨语言
序列化后的体积太大，是二进制编码的 5 倍多
序列化性能太低

Protobuf

Protobuf 是 Google 发布的开源项目，全称 Google Protocol Buffers，是一种轻便高效的结构化数据存储格式，可以用于结构化数据串行化，或者说序列化。它很适合做数据存储或 RPC(远程过程调用 remote procedure call)数据交换格式。Protobuf 是以 message 的方式来管理数据的。支持跨平台、跨语言，即客户端和服务器端可以是不同的语言编写的（支持目前绝大多数语言，例如 C++、C#、Java、python 等）

Protobuf使用示例

1.idea安装proto插件

2.下载proto.exe(windows上使用，注意版本)

protoc-3.12.1-win64.zip

3.引入依赖

xml

<dependency>
<groupId>com.google.protobuf</groupId>
	<artifactId>protobuf-java</artifactId>
	<version>3.12.2</version>
</dependency>

4.编写proto文件

protobuf类型对照

java

syntax = "proto3";  //协议的版本
option java_outer_classname = "StudentPOJO";  //生成的外部类名，也是文件名
//protobuf 使用 message 管理数据
message Student { //会在 StudentPOJO 外部类生成一个内部类，名称为Student，它是真正发送的pojo对象
  int32 id = 1; //Student类中，有1个属性，名称为id，类型为int32(protobuf类型)，1表示属性序号，不是值
  string name = 2;
}

5.生成java实体类

将下载的protoc-3.12.1-win64解压，将上面编写的proto文件复制到解压后的bin目录，与proto.exe同级 -> 在当前目录执行cmd命令，执行后会在该目录生成一个java实体类。将实体类再复制到项目中。

shell

.\protoc.exe --java_out=. .\Student.proto

生成的StudentPOJO.java： 点击下载 StudentPOJO.java

6.使用示例

java

//客户端加入 handler，加入编码器
ChannelPipeline pieline = ch.pipeline();
//加入 Protobuf 编码器
pieline.addLast(new ProtobufEncoder());
pieline.addLast(new NettyClientHandler());


//服务端加入 handler，加入解码器
ChannelPipeline pieline = ch.pipeline();
//加入 Protobuf 解码器，需指定对哪种对象进行解码
pieline.addLast(new ProtobufDecoder(StudentPOJO.Student.getDefaultInstance()));
pieline.addLast(new NettyServerHandler());


//客户端发送消息给服务端，发送一个Student对象到服务器
StudentPOJO.Student student = StudentPOJO.Student.newBuilder().setId(5).setName("张三").build();
ctx.writeAndFlush(student);


//服务端接收student
//读取从客户端发送的StudentPOJO.Student
StudentPOJO.Student student = (StudentPOJO.Student) msg;
int id = student.getId();
String name = student.getName();

Protobuf传输多种类型使用示例

1.修改Student.proto，新增1个Worker类型

java

syntax = "proto3";
option optimize_for = SPEED;    //加快解析
option java_outer_classname = "MyDataInfo";   //外部类名称
//protobuf 可以使用message管理其他的message
message MyMessage{
    //定义一个枚举
    enum DataType{
        StudentType = 0;    //proto3要求enum的编号从0开始
        WorkerType = 1;
    }

    DataType data_type = 1;//用data_type来标识传的是哪一个枚举类型，它是MyMessage的一个属性
    //表示每次枚举类型最多只能出现 Student或Worker中的一个
    oneof dateBody {
        //这里因为data_type为1，所以依次序号为2，3。
        //这个oneof表示，MyMessage中的属性，包含data_type和dateBody，
        //而dateBody中要么出现Student，要么出现Worker，只能出现一个。有多个同理。
        Student student = 2;
        Worker worker = 3;
    }

}
message Student {
    int32 id = 1;   //Student类属性
    string name = 2;
}
message Worker {
    string name = 1;    //work类属性
    int32 age = 2;
}

2.生成对应的java类

点击下载 MyDataInfo.java

3.发送消息

java

//客户端加入 handler，加入编码器
ChannelPipeline pieline = ch.pipeline();
//加入 Protobuf 编码器
pieline.addLast(new ProtobufEncoder());
pieline.addLast(new NettyClientHandler());


//服务端加入 handler，加入解码器
ChannelPipeline pieline = ch.pipeline();
//加入 Protobuf 解码器，需指定对哪种对象进行解码
pieline.addLast(new ProtobufDecoder(MyDataInfo.MyMessage.getDefaultInstance()));
pieline.addLast(new NettyServerHandler());


//随机发送Student或Worker
int random = new Random().nextInt(3);
MyDataInfo.MyMessage myMessage = null;
if(random == 0){
    myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.StudentType)
            .setStudent(MyDataInfo.Student.newBuilder().setId(5).setName("张三").build()).build();
} else {
    myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.WorkerType)
            .setWorker(MyDataInfo.Worker.newBuilder().setAge(25).setName("李四").build()).build();
}
ctx.writeAndFlush(myMessage);


//服务端接收 MyDataInfo.MyMessage
MyDataInfo.MyMessage myMessage = (MyDataInfo.MyMessage) msg;
//根据 DataType 获取不同的信息
MyDataInfo.MyMessage.DataType dataType = myMessage.getDataType();
if(dataType == MyDataInfo.MyMessage.DataType.StudentType){
    MyDataInfo.Student student = myMessage.getStudent();
    System.out.println("学生id：" + student.getId() + " - 学生姓名：" + student.getName());
} else if(dataType == MyDataInfo.MyMessage.DataType.WorkerType){
    MyDataInfo.Worker worker = myMessage.getWorker();
    System.out.println("学生姓名：" + worker.getName() + " - 学生年龄：" + worker.getAge());
} else {
    System.out.println("类型不正确！");
}

Netty快速入门 ​

Netty是什么 ​

原生NIO存在的问题 ​

Netty的优点 ​

Netty版本说明 ​

线程模型基本介绍 ​

传统阻塞I/O服务模型 ​

Reactor 模式 ​

Reactor 模式中核心组成 ​

单 Reactor 单线程 ​

单Reactor多线程 ​

主从 Reactor 多线程 ​

Netty模型 ​

Netty代码示例 - TCP服务 ​

Netty服务端 - Server ​

Netty服务端 - Handler ​

Netty客户端 - Client ​

Netty客户端 - Handler ​

Netty代码示例 - HTTP服务 ​

异步模型 ​

taskQueue自定义任务 ​

scheduleTaskQueue自定义定时任务 ​

Bootstrap、ServerBootstrap ​

Future、ChannelFuture ​

Future-Listener 机制 ​

Channel ​

ChannelHandler ​

出站/入站 ​

Pipeline 和 ChannelPipeline ​

ChannelHandlerContext ctx ​

option和childOption ​

EventLoopGroup 和其实现类 NioEventLoopGroup ​

Unpooled 类 ​

Netty代码示例 - 消息群聊 ​

群聊服务端 - Server ​

群聊服务端 - handler ​

群聊客户端 - Client ​

群聊客户端 - handler ​

心跳监测 ​

TimeoutHandler ​

Netty代码示例 - WebSocket长连接 ​

WebSocket长连接服务端 - Server ​

WebSocket长连接 - handler ​

WebSocket长连接 - 页面调用 ​

编码和解码 ​

Netty 本身的编码解码的机制和问题 ​

Protobuf ​

Protobuf使用示例 ​

1.idea安装proto插件 ​

2.下载proto.exe(windows上使用，注意版本) ​

3.引入依赖 ​

4.编写proto文件 ​

5.生成java实体类 ​

6.使用示例 ​

Protobuf传输多种类型使用示例 ​

1.修改Student.proto，新增1个Worker类型 ​

2.生成对应的java类 ​

3.发送消息 ​

Netty快速入门

Netty是什么

原生NIO存在的问题

Netty的优点

Netty版本说明

线程模型基本介绍

传统阻塞I/O服务模型

Reactor 模式

Reactor 模式中核心组成

单 Reactor 单线程

单Reactor多线程

主从 Reactor 多线程

Netty模型

Netty代码示例 - TCP服务

Netty服务端 - Server

Netty服务端 - Handler

Netty客户端 - Client

Netty客户端 - Handler

Netty代码示例 - HTTP服务

异步模型

taskQueue自定义任务

scheduleTaskQueue自定义定时任务

Bootstrap、ServerBootstrap

Future、ChannelFuture

Future-Listener 机制

Channel

ChannelHandler

出站/入站

Pipeline 和 ChannelPipeline

ChannelHandlerContext ctx

option和childOption

EventLoopGroup 和其实现类 NioEventLoopGroup

Unpooled 类

Netty代码示例 - 消息群聊

群聊服务端 - Server

群聊服务端 - handler

群聊客户端 - Client

群聊客户端 - handler

心跳监测

TimeoutHandler

Netty代码示例 - WebSocket长连接

WebSocket长连接服务端 - Server

WebSocket长连接 - handler

WebSocket长连接 - 页面调用

编码和解码

Netty 本身的编码解码的机制和问题

Protobuf

Protobuf使用示例

1.idea安装proto插件

2.下载proto.exe(windows上使用，注意版本)

3.引入依赖

4.编写proto文件

5.生成java实体类

6.使用示例

Protobuf传输多种类型使用示例

1.修改Student.proto，新增1个Worker类型

2.生成对应的java类

3.发送消息