一文搞定Netty，打造单机百万连接测试！1

public class BioServer {
    private static final int PORT = 8080;
    public static void main(String[] args) throws IOException {
        //新建socketServer
        ServerSocket serverSocket = null;
        try{
            //绑定对应端口
            serverSocket = new ServerSocket(PORT);
            System.out.println("the time server is start in port :"+PORT);
            Socket socket = null;
            while(true){
                //拿到请求进来的socket
                socket = serverSocket.accept();
                //线程请求
                new Thread(new TimeServerHandler(socket)).start();
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if (serverSocket != null){
                System.out.println("the time server close");
                serverSocket.close();
            }
        }
    }
}

public class TimeServerHandler implements Runnable{
    private Socket socket;
    public TimeServerHandler(Socket socket) {
        this.socket = socket;
    }
    public TimeServerHandler() {
    }
    @Override
    public void run() {
        BufferedReader in = null;
        PrintWriter out = null;
        try {
            in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
            //为true时autoFlush自动刷新，无需在调用flush方法
            out = new PrintWriter(this.socket.getOutputStream(),true);
            String body  = null;
      //循环监听客户端发送的msg
            while ((body = in.readLine())!=null && body.length()!=0){
                System.out.println("this time server receive msg :"+body);
                out.println(new Date().toString());
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(in != null){
                try {
                    in.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
            if(out != null){
                try {
                    out.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
            if(this.socket != null){
                try {
                    this.socket.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }
}

public class BioClient {
    private static final String HOST = "127.0.0.1";
    private static final int PORT = 8080;
    public static void main(String[] args) {
        Socket socket = null;
        BufferedReader in = null;
        PrintWriter out = null;
        try{
            //创建连接
            socket = new Socket(HOST,PORT);
            in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            out = new PrintWriter(socket.getOutputStream(),true);
            out.println("I am client");
            String resp = in.readLine();
            System.out.println("当前服务器时间是："+resp);
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(in != null){
                try {
                    in.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
            if(out != null){
                try {
                    out.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
            if(socket != null){
                try {
                    socket.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }
}

第一阶段：等待数据准备（Waiting for the data to be ready）
第二阶段：将数据从内核拷贝到进程中（Copy the data from the kernel to the process）

int select(int maxfdp1,fd_set *readset,fd_set *writeset,fd_set *exceptset,const struct timeval *timeout);

select()提供了一种fd_set的核心数据结构，实际上是一个long类型的数组，每一个数组元素都能与一个打开的文件句柄（不管是Socket句柄，还是其他文件/命令管道/设备句柄）建立联系。当调用select()时，由内核根据IO状态修改fd_set的内容，由此来通知执行了select()的进程哪一个socket或文件可读。

一般来说这个数目和系统的内存关系很大，具体数目可以cat /proc/sys/fs/file-max 查看，32位默认是1024，64位默认是2048

当套接字比较多的时候，每次select()都要通过遍历FD_SETSIZE个Socket来完成调度，不管哪个Socket是活跃的，都要遍历一遍，这会浪费很多CPU时间。

int poll(struct pollfd *fds,nfds_t nfds,int timeout);
typedef struct pollfd{
  int fd;     //需要被检测或选择的文件描述符
  short events;   //对文件描述符fd上感兴趣的事件
  short revents;  //文件描述符fd上当前实际发生的事件
} pollfd_t;

1、大量的fd的数组被整体复制于用户态和内核地址空间之间，而不管这样的复制是不是有意义的。
2、poll还有一个特点是“水平触发”，如果报告了fd后，没有被处理，那么下次poll时会再次报告该fd。

int epoll_create(int size);
int epoll_ctl(int epfd,int op,int fd,struct epoll_event *event);
int epoll_wait(int epfd,struct epoll_event *events,int maxevents,int timeout);

struct eventpoll{
    ...
    //红黑树的根节点，树中存储这所有添加到epoll中需要被监听的事件
    struct rb_root rbr;
    //双链表，存放这通过epoll_wait()返回的就绪事件
    struct list_head rdlist;
}

每个epoll对象都有一个独立的eventpoll，用于存放通过epoll_ctl()添加进来的事件。这些事件维护在红黑树中，红黑树的插入时间效率是log(n)（n为树的高度）

此外，被监听的事件都会与设备驱动程序建立回调关系，每当被监听的事件就绪，系统注册的回调函数就会被调用，将就绪事件放到rdList中，时间复杂度为O(1).
当调用epoll_wait()时，无需遍历整个被监听的描述符集，只需要遍历eventpoll对象中的rdlist双链表中是否有epitem元素即可。然后就把就绪事件复制到用户态，同时将事件数量返回给用户。

struct epitem{
    struct rb_node  rbn;         // 红黑树节点
    struct list_head    rdllink; // 双向链表节点
    struct epoll_filefd  ffd;    // 事件句柄信息
    struct eventpoll *ep;        // 指向所属的eventpoll对象
    struct epoll_event event;    // 期待发生的事件类型
}

1、jdk1.4之前是采用同步阻塞模型，也就是BIO
    大型服务一般采用C或者C++, 因为可以直接操作系统提供的异步IO,AIO
2、jdk1.4推出NIO,支持非阻塞IO，jdk1.7升级,推出NIO2.0,提供AIO的功能，支持文件和网络套接字的异步IO

设计模式——Reactor模式（反应器设计模式），是一种基于事件驱动的设计模式，在事件驱动的应用中，将一个或多个客户的服务请求分离（demultiplex）和调度（dispatch）给应用程序。在事件驱动的应用中，同步地、有序地处理同时接收的多个服务请求
一般出现在高并发系统中，比如Netty，Redis等

<!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.32.Final</version>
        </dependency>

public class EchoServer{
    //设定端口号
    private int port;
    //构造方法传入端口
    public EchoServer(int port){
        this.port = port;
    }
    public void run() throws InterruptedException {
        //创建线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try{
          //创建启动引导类
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            //设置线程组
            serverBootstrap.group(bossGroup,workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>(){
                    protected void initChannel(SocketChannel socketChannel) throws Exception{
                        //设置处理器，可以设置多个
                        socketChannel.pipeline().addLast(new EchoServerHandler());
                    }
                });
            System.out.println("Echo服务启动中...");
            //绑定端口，同步等待
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            //等待服务端监听端口关闭
            channelFuture.channel().closeFuture().sync();
        }finally{
            //优雅退出
            workGroup.shutdownGracefully();
            bossGroup.shutdownGracefullt();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        //设置默认的端口
        int port = 8080;
        if(args.length>0){
            port = Integer.parseInt(args[0]);
        }    
        //调用启动方法
        new EchoServer(port).run();
    }
}

public EchoServerHandler extends ChannelInboundHandlerAdapter{
    @Override
    public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception{
        ByteBuf data = (ByteBuf) msg;
        System.out.println("服务端收到数据："+data.toString(CharsetUtil.UTF_8));
        //注意：数据一定要回写出去，不然客户端收不到
        ctx.writeAndFlush(data);
    } 
   @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("EchoServerHandler EchoServerHandler()");
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

public class EchoClient{
    //客户端发起请求的ip地址
    private String host;
    //端口
    private int port;
    //构造方法传入ip+端口初始化
    public EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }
    public void start(){
        //创建线程组
        EventLoopGroup group = new NioEventLoopGroup();
        try{
            //创建启动引导类
            BootStrap bootStrap = new BootStrap();
            bootStrap.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(host,port))
                .handler(new ChannelInitializer<SocketChannel>{
               @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //设置客户端处理器
                            socketChannel.pipeline().addLast(new EchoClientHandler());
                        }   
            });
            //异步连接，同步阻塞，connect是异步连接
            ChannelFuture channelFuture = bootStrap.connect().sync();
            //阻塞住直到客户端通道关闭
            channelFuture.channel().closeFuture().sync();
        }finally{
            //优雅退出
            group.shutdownGracefully();
        }
    }
    public static void main(String[] args){
        //启动客户端连接
        new EchoClient("127.0.0.1",8080).start();
    } 
}

public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
        System.out.println("Client received："+byteBuf.toString(CharsetUtil.UTF_8));
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("EchoClientHandler.channelActive()");
        //创建一个缓存，使用Unpooled工具类
        ctx.writeAndFlush(Unpooled.copiedBuffer("李祥",CharsetUtil.UTF_8));
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("EchoClientHandler.channelReadComplete()");
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

channel是客户端和服务端建立的一个连接，是一个socket连接，具有生命周期，建立成功，读取数据，读取完成，出现异常等等。

channelHandler是做处理的，对接收的数据进行处理到要直接或间接继承channelHandler。channelPipeline就好比一个处理工厂，可以添加很多handler处理类，当进入channelHandler后都要经过pipeline添加的handler进行处理。