【netty】

一、概述

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

核心：可扩展事件模型、api库、零拷贝

协议支持：http、websocket、ssl（安全套接字协议）、谷歌protobuf、zlib/gzip压缩与解压、large file transfer大文件传输

传输服务：socket、datagram、http tunnel

二、特点

1、netty优点

API使用简单，学习成本低

功能强大，内置了多种解、编码器

性能高，对比其他主流NIO框架，netty性能最优

社区活跃

dubbo、Elasticsearch 都采用了netty，质量得到验证

2、NIO缺点

NIO类库和API繁杂，（Selector、ServerSocketChanne、SocketChannel、ByteBuffer）

对Java多线程理解要高

epoll bug，导致selector空轮询，导致cpu100%

三、helloworld

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 DiscardServer {

    private int port;

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

    public void run() throws InterruptedException {
        //线程组
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();

        try {
            //服务端启动对象
            ServerBootstrap b = new ServerBootstrap();
            //设置上面创建好的线程组
            b.group(bossGroup,workGroup)
                    //设置服务端通道的实现类型
                    .channel(NioServerSocketChannel.class)
                    //使用匿名内部类的形式初始化通道对象
                    .childHandler(new ChannelInitializer() {
                        //给workergroup的eventloop对应的管道设置处理器
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //给pipeline管道设置处理器
                            ch.pipeline().addLast(new DiscardServerHandler());
                        }
                    })
                    //设置线程队列得到链接个数
                    .option(ChannelOption.SO_BACKLOG,128)
                    //设置保持活动的链接状态
                    .childOption(ChannelOption.SO_KEEPALIVE,true);
            //绑定端口好，启动服务端
            ChannelFuture future = b.bind(port).sync();
            //对关闭管道进行监听
            future.channel().closeFuture().sync();

        }finally {
            workGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        int port = 8080;
        if (args.length > 0){
            port = Integer.parseInt(args[0]);
        }
        new DiscardServer(port).run();
    }
}



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



public class DiscardServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //通常msg逻辑，finally 释放
        //获取客户端发送来的消息
       ByteBuf in =  (ByteBuf) msg;
        try {
            while (in.isReadable()){
                //丢弃服务器
//                System.out.println(in.toString());
//                System.out.flush();
                // 回声服务器
                ctx.write(msg);
                ctx.flush();
            }

        }finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //发送消息到客户端
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello", CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //异常关闭
        cause.printStackTrace();
        ctx.close();
    }
}


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

public class Timeclient {

    public static void main(String[] args) throws InterruptedException {
        String host = "127.0.0.1";
        int port = 8080;
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            //类比服务端：创建bootstrap 配置参数
            Bootstrap b = new Bootstrap();
            //设置线程组
            b.group(workGroup)
                    //设置客户端通道实现类型
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    //使用匿名内部类初始化通道
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //给pipeline添加处理器
                            ch.pipeline().addLast(new TimeClientHandler());
                        }
                    });
            //链接服务端
            ChannelFuture future = b.connect(host, port);
            //对通道关闭进行监听
            future.channel().closeFuture().sync();
        } finally {
            //关闭线程组
            workGroup.shutdownGracefully();
        }
    }
}

四、组件

1、taskQueue任务队列

2、scheduleTaskQueue延时任务队列

3、future异步机制

4、Bootstrap ServerBootStrap extends AbstractBootstrap

创建启动器过程：

1、设置eventLoopGroup

2、设置channel（new NioSocketChannel.class）

3、设置option（ChannelOption）

4、是在handler

5、bind端口

6、启动

7、等待通道关闭

8、优雅关闭 eventLoopGroup

5、group

服务端一般需要两个线程组

bossGroup：监听客户端连接，负责与客户端创建连接，把连接注册到workerGroup中的Selector。

workerGroup：处理每个连接的读写事件

两个group创建直接new NioEventLoopGroup();

默认线程数：

Math.max(1, SystemPropertyUtil.getInt( "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2)

6、channel 通道类型

1、NioSocketChannel:异步非阻塞的客户端TCP Socket连接

2、NioSeverSocketChannel:异步非阻塞的服务器端TCP Socket连接

3、OioSocketChannel：同步阻塞客户端

4、OioServerSocketChannel：服务器

7、option() 与 childoption()

option ：设置的是 服务端用于接受进来的连接，bossgroup 的进程

childoption：提供给 父管道接收的连接，workergrouop 线程

option() 参数： serverSocketChannel :

SO_BACKLOG(win:200，其他：128)：服务器接受连接的队列长度，队列满，在客户端连接被拒绝。

childoption() 参数：SocketChannel

SO_RECBUF：socket参数，tcp数据接收的缓冲区大小

TCP_NODELAY：tcp参数，立即发送数据，默认true

SO_KEEPLIVE ：socket参数，连接保活，默认false，启用该功能，tcp主动探测空闲连接有效性

8、设置流水线 pipeline

ChannelPipeline：netty处理请求的责任链，channelHandler 具体处理请求的处理器。

bootStrap.childHandler(new ChannelInitializer )初始化通道，装配流水线从这个地方开始

handler处理器：ChannnelInboundHandlerAdapter、ChannelOutboundHandlerAdapter

in（入站处理器）：数据从JavaNIO到nettychannel

out（出战）：相反

in常用事件（接口）：

注册事件 ：channelRegistered

连接建立事件：channelActive

读事件和读完成事件：channelRead、channelReadComplete

异常通知事件：ExceptionCaught

用户自定义事件：userEventTriggered

channel可写状态变化事件：channelWritabilitychanged

连接关闭事件：channelInactive

out处理器常用事件

端口绑定：bind

连接服务端：connect

写事件：write

刷新事件：flush

读事件：read

主动断开连接：disconnect

关闭channel事件：close

9、bind

提供用于服务端和客户端绑定服务器地址和端口，默认异步启动；sync()同步。

10、优雅关闭EventLoopGroup

shutdownGracefully();关闭所以child Channel 释放底层资源

11、channel

组件：连接到网络套接字 或者能进行读写连接绑定io

1、获取channel状态

2、获取channel的配置参数

channel.config().getOption channel.config().getOptions

3、channel的IO操作

写操作：ctx.channel().writeAndFlush

连接操作：channelFuture.channel().connect()

通过channel获取ChannelPipeline：ctx.channel().pipeline();

12、Selector

用于监听事件，管理注册到其中到channnel，实现多路复用

13、pipeline 与 channelPipeline

pipeline：处理器的容器。

一个channel，只有一个channelPipeline，在channel创建的时候被创建

channelPipeline包含一个ChannelHandler形成的列表，所以的channelHandler都会被注册到ChannelPipeline中。

14、ChannelHandlerContext

通过这个获取pipeline 、channel对象。

15、EventLoopGroup

每个服务器需要两个线程组配合工作，线程组接口就是EventLoopGroup。

每个group包含一个或多个EventLoop ，每个loop维护一个Selector实例。

五、轮询机制

NIO

一、Java.nio（new IO）为所有原始类型（除boolean）提供缓存支持的数据容器。