一文带你由浅入深Netty异步非阻塞世界

package com.anxpp.io.calculator.bio;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
/**
 * BIO服务端源码
 * @author shenj
 * @version 1.0
 */
public final class ServerNormal {
  //默认的端口号
  private static int DEFAULT_PORT = 12345;
  //单例的ServerSocket
  private static ServerSocket server;
  //根据传入参数设置监听端口，如果没有参数调用以下方法并使用默认值
  public static void start() throws IOException{
    //使用默认值
    start(DEFAULT_PORT);
  }
  //这个方法不会被大量并发访问，不太需要考虑效率，直接进行方法同步就行了
  public synchronized static void start(int port) throws IOException{
    if(server != null) return;
    try{
      //通过构造函数创建ServerSocket
      //如果端口合法且空闲，服务端就监听成功
      server = new ServerSocket(port);
      System.out.println("服务器已启动，端口号：" + port);
      //通过无线循环监听客户端连接
      //如果没有客户端接入，将阻塞在accept操作上。
      while(true){
        Socket socket = server.accept();
        //当有新的客户端接入时，会执行下面的代码
        //然后创建一个新的线程处理这条Socket链路
        new Thread(new ServerHandler(socket)).start();
      }
    }finally{
      //一些必要的清理工作
      if(server != null){
        System.out.println("服务器已关闭。");
        server.close();
        server = null;
      }
    }
  }
}

package com.anxpp.io.calculator.bio;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import com.anxpp.io.utils.Calculator;
/**
 * 客户端线程
 * @author shenj
 * 用于处理一个客户端的Socket链路
 */
public class ServerHandler implements Runnable{
  private Socket socket;
  public ServerHandler(Socket socket) {
    this.socket = socket;
  }
  @Override
  public void run() {
    BufferedReader in = null;
    PrintWriter out = null;
    try{
      in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
      out = new PrintWriter(socket.getOutputStream(),true);
      String expression;
      String result;
      while(true){
        //通过BufferedReader读取一行
        //如果已经读到输入流尾部，返回null,退出循环
        //如果得到非空值，就尝试计算结果并返回
        if((expression = in.readLine())==null) break;
        System.out.println("服务器收到消息：" + expression);
        try{
          result = Calculator.cal(expression).toString();
        }catch(Exception e){
          result = "计算错误：" + e.getMessage();
        }
        out.println(result);
      }
    }catch(Exception e){
      e.printStackTrace();
    }finally{
      //一些必要的清理工作
      if(in != null){
        try {
          in.close();
        } catch (IOException e) {
          e.printStackTrace();
        }
        in = null;
      }
      if(out != null){
        out.close();
        out = null;
      }
      if(socket != null){
        try {
          socket.close();
        } catch (IOException e) {
          e.printStackTrace();
        }
        socket = null;
      }
    }
  }
}

package com.anxpp.io.calculator.bio;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
/**
 * 阻塞式I/O创建的客户端
 * @author shenj
 * @version 1.0
 */
public class Client {
  //默认的端口号
  private static int DEFAULT_SERVER_PORT = 12345;
  private static String DEFAULT_SERVER_IP = "127.0.0.1";
  public static void send(String expression){
    send(DEFAULT_SERVER_PORT,expression);
  }
  public static void send(int port,String expression){
    System.out.println("算术表达式为：" + expression);
    Socket socket = null;
    BufferedReader in = null;
    PrintWriter out = null;
    try{
      socket = new Socket(DEFAULT_SERVER_IP,port);
      in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
      out = new PrintWriter(socket.getOutputStream(),true);
      out.println(expression);
      System.out.println("___结果为：" + in.readLine());
    }catch(Exception e){
      e.printStackTrace();
    }finally{
      //一下必要的清理工作
      if(in != null){
        try {
          in.close();
        } catch (IOException e) {
          e.printStackTrace();
        }
        in = null;
      }
      if(out != null){
        out.close();
        out = null;
      }
      if(socket != null){
        try {
          socket.close();
        } catch (IOException e) {
          e.printStackTrace();
        }
        socket = null;
      }
    }
  }
}

package com.anxpp.io.calculator.bio;
import java.io.IOException;
import java.util.Random;
/**
 * 测试方法
 * @author shenj
 * @version 1.0
 */
public class Test {
  //测试主方法
  public static void main(String[] args) throws InterruptedException {
    //运行服务器
    new Thread(new Runnable() {
      @Override
      public void run() {
        try {
          ServerBetter.start();
        } catch (IOException e) {
          e.printStackTrace();
        }
      }
    }).start();
    //避免客户端先于服务器启动前执行代码
    Thread.sleep(100);
    //运行客户端 
    char operators[] = {'+','-','*','/'};
    Random random = new Random(System.currentTimeMillis());
    new Thread(new Runnable() {
      @SuppressWarnings("static-access")
      @Override
      public void run() {
        while(true){
          //随机产生算术表达式
          String expression = random.nextInt(10)+""+operators[random.nextInt(4)]+(random.nextInt(10)+1);
          Client.send(expression);
          try {
            Thread.currentThread().sleep(random.nextInt(1000));
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
      }
    }).start();
  }
}

1)每个 channel都会对应一个 Buffer
2)Selector对应一个线程，一个线程对应多个 channel（连接）
3)该图反应了有三个 channel？注册到该 selector/程序
4)程序切换到哪个 channel是有事件决定的， Event就是一个重要的概念
5)Selector会根据不同的事件，在各个通道上切换
6)Buffer就是一个内存块，底层是有一个数组
7)数据的读取写入是通过 Buffer，这个和BIO,BIO中要么是输入流，或者是
输出流，不能双向，但是NIO的 Buffer是可以读也可以写，需要ip方法切换
8)channel是双向的，可以返回底层操作系统的情况，比如 Linux，底层的操作系
通道就是双向的

package icu.lookyousmileface.nio.basic;
import java.nio.IntBuffer;
/**
 * @author shenj
 * @title: NioBuffer
 * @projectName NettyPro
 * @date 2020/11/30 14:41
 */
public class NioBuffer {
    public static void main(String[] args) {
        //IntBuffer.allocate(5);表示Buffer的空间为5，并且Buffer缓冲区类型为Int
        IntBuffer intBuffer = IntBuffer.allocate(5);
        //intBuffer.capacity()表示获得Buffer的大小
        for(int i = 0;i< intBuffer.capacity();i++){
            intBuffer.put(i*2);
        }
        //buffer进行过写操作之后需要读操作的时候需要flip进行状态反转
        intBuffer.flip();
        //hasRemaining()返回剩余的可用长度
        while (intBuffer.hasRemaining()){
            System.out.println(intBuffer.get());
        }
    }
}

/**
 * @author shenj
 * @title: NioBufferChannelFileWrite
 * @projectName NettyPro
 * @date 2020/11/30 18:05
 */
public class NioBufferChannelFileWrite {
    private static final String msg = "怒发冲冠，凭栏处、潇潇雨歇。抬望眼、仰天长啸，壮怀激烈。三十功名尘与土，八千里路云和月。莫等闲、白了少年头，空悲切。";
    public static void main(String[] args) {
        try {
            FileOutputStream fileOutputStream = new FileOutputStream(new File("src/main/resources/filedata/data1.txt"));
            FileChannel fileOutputStreamChannel = fileOutputStream.getChannel();
            ByteBuffer fileDataBuffer = ByteBuffer.allocate(1024);
            ByteBuffer putData = fileDataBuffer.put(msg.getBytes());
            //反转
            putData.flip();
            fileOutputStreamChannel.write(putData);
            fileOutputStream.close();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

/**
 * @author shenj
 * @title: NioBufferChannelRead
 * @projectName NettyPro
 * @date 2020/11/30 18:55
 */
public class NioBufferChannelRead {
    private static final String filePath = "src/main/resources/filedata/data1.txt";
    public static void main(String[] args) {
        File file = new File(filePath);
        try {
            FileInputStream fileInputStream = new FileInputStream(file);
            FileChannel fileInputStreamChannel = fileInputStream.getChannel();
            //获取file的大小，避免浪费内存
            ByteBuffer byteDataBuffer = ByteBuffer.allocate((int) file.length());
            fileInputStreamChannel.read(byteDataBuffer);
            //byteDataBuffer.array()将byteBuffer缓冲区中的data变成数组
            System.out.println(new String(byteDataBuffer.array()));
            fileInputStream.close();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

/**
 * @title: NioScatingAndGething
 * @projectName NettyPro
 * @author shenj
 * @date 2020/12/1 13:23
 */
fun main(args: Array<String>): Unit {
    val serverSocketChannel = ServerSocketChannel.open();
    val inetSocketAddress = InetSocketAddress(9948);
    //绑定端口到socket上，并启动
    serverSocketChannel.socket().bind(inetSocketAddress);
    //buffer数组,NIO会自动将数据放到数组中，无需操心
    val byteBuffer = arrayOfNulls<ByteBuffer>(2)
    byteBuffer[0] = ByteBuffer.allocate(5)
    byteBuffer[1] = ByteBuffer.allocate(3)
    val scoketChannel = serverSocketChannel.accept();
    var messageLight = 8;
    while (true) {
        var byteRead: Int = 0
        while (byteRead < messageLight) {
            var read = scoketChannel.read(byteBuffer)
            println("byteRead:" + read)
            byteRead += read.toInt()
            Arrays.asList<ByteBuffer>(*byteBuffer).stream().map { buffer: ByteBuffer -> "potion:" + buffer.position() + "limit:" + buffer.limit() }.forEach { x: String? -> println(x) }
        }
        Arrays.asList<ByteBuffer>(*byteBuffer).forEach(Consumer { buffer: ByteBuffer -> buffer.flip() })
        var byteWrite: Long = 0;
        while (byteWrite < messageLight) {
            var write = scoketChannel.write(byteBuffer)
            byteWrite += write
        }
        Arrays.asList<ByteBuffer>(*byteBuffer).forEach(Consumer { buffer: ByteBuffer -> buffer.clear() })
        println("byteRead:" + byteRead + "byteWrite:" + byteWrite + "messagelenght:" + messageLight)
    }
}

/**
 * @author shenj
 * @title: NioBufferOnlyOneWriteAndRead
 * @projectName NettyPro
 * @date 2020/11/30 19:18
 */
public class NioBufferOnlyOneWriteAndRead {
    private static final String filePath1 = "src/main/resources/filedata/data1.txt";
    private static final String filePath2 = "src/main/resources/filedata/data2.txt";
    public static void main(String[] args) {
        File file1 = new File(filePath1);
        File file2 = new File(filePath2);
        try {
            FileInputStream fileInputStream = new FileInputStream(file1);
            FileOutputStream fileOutputStream = new FileOutputStream(file2);
            FileChannel fileInputStreamChannel = fileInputStream.getChannel();
            FileChannel fileOutputStreamChannel = fileOutputStream.getChannel();
            ByteBuffer dataBuffer = ByteBuffer.allocate(512);
            while (-1 != fileInputStreamChannel.read(dataBuffer)) {
                //读和写之间需要切换
                dataBuffer.flip();
                fileOutputStreamChannel.write(dataBuffer);
                //清空Buffer缓冲区的数据
                dataBuffer.clear();
            }
            fileInputStream.close();
            fileOutputStream.close();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

/**
 * @author shenj
 * @title: NioBufferTransorf
 * @projectName NettyPro
 * @date 2020/11/30 20:18
 */
public class NioBufferTransorf {
    private static final String filepath1 = "src/main/resources/filedata/data1.txt";
    private static final String filepath2 = "src/main/resources/filedata/data2.txt";
    public static void main(String[] args) {
        try {
            FileInputStream fileInputStream = new FileInputStream(new File(filepath1));
            FileOutputStream fileOutputStream = new FileOutputStream(new File(filepath2));
            FileChannel fileInputStreamChannel = fileInputStream.getChannel();
            FileChannel fileOutputStreamChannel = fileOutputStream.getChannel();
            //将目标的通道的数据复制到当前通道,Channel自带数据有效长度size获取
            fileOutputStreamChannel.transferFrom(fileInputStreamChannel, 0, fileInputStreamChannel.size());
            fileInputStreamChannel.close();
            fileOutputStreamChannel.close();
            fileInputStream.close();
            fileOutputStream.close();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

fun main(args: Array<String>):Unit {
    val  imagePath = "src/main/resources/filedata/sky.jpg";
    val  copyImageToPath = "src/main/resources/filedata/sky_copy.jpg"
    val accessFile = RandomAccessFile(imagePath, "rw")
    val accessFile_copy = RandomAccessFile(copyImageToPath, "rw")
    val accessFile_channle = accessFile.channel
    val accesssFile_copy_channel = accessFile_copy.channel
    accesssFile_copy_channel.transferFrom(accessFile_channle,0,accessFile_channle.size())
}

package bigdata.studynio;
public class NioServer {
  public static void main(String[] args) {
    int port = 8080;
    if(args != null && args.length < 0){
      //port = Integer.valueOf(args[0]);  
    }
    //创建服务器线程
    NioServerWork nioServerWork = new NioServerWork(port);
    new Thread(nioServerWork, "server").start();
  }
}

package bigdata.studynio;
import java.io.BufferedReader;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class NioServerWork implements Runnable {
  //多路复用器 Selector会对注册在其上面的channel进行；轮询，当某个channel发生读写操作时，
  //就会处于相应的就绪状态，通过SelectionKey的值急性IO 操作
  private Selector selector;//多路复用器
  private ServerSocketChannel channel;
  private volatile boolean stop;
  /**
   * @param port
   * 构造函数
   */
  public NioServerWork(int port) {
    try {
      selector = Selector.open();//打开多路复用器
      channel = ServerSocketChannel.open();//打开socketchannel
      channel.configureBlocking(false);//配置通道为非阻塞的状态
      channel.socket().bind(new InetSocketAddress(port), 1024);//通道socket绑定地址和端口
      channel.register(selector, SelectionKey.OP_ACCEPT);//将通道channel在多路复用器selector上注册为接收操作
      System.out.println("NIO 服务启动 端口： "+ port);
    } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }   
  }
  public void stop(){
    this.stop=true;
  }
  @Override
  public void run() {//线程的Runnable程序
    System.out.println("NIO 服务  run()");
    while(!stop){
      try {
        selector.select(1000);//最大阻塞时间1s
        //获取多路复用器的事件值SelectionKey，并存放在迭代器中
        Set<SelectionKey> selectedKeys = selector.selectedKeys();
        Iterator<SelectionKey> iterator = selectedKeys.iterator();
        SelectionKey key =null;
        //System.out.println("NIO 服务  try");
        while(iterator.hasNext()){
          System.out.println("NIO 服务  iterator.hasNext()");
          key = iterator.next();
          iterator.remove();//获取后冲迭代器中删除此值
          try {
            handleinput(key);//根据SelectionKey的值进行相应的读写操作        
          } catch (Exception e) {
            if(key!=null){
              key.cancel();
              if(key.channel()!=null)
                key.channel().close();              
            }
          }                 
        }             
      } catch (IOException e) {
        System.out.println("NIO 服务  run  catch IOException");
        e.printStackTrace();
        System.exit(1);
      }
    }   
  }
  /**
   * @param key
   * @throws IOException
   * 根据SelectionKey的值进行相应的读写操作
   */
  private void handleinput(SelectionKey key) throws IOException {
    System.out.println("NIO 服务  handleinput");
    if(key.isValid()){//判断所传的SelectionKey值是否可用
      if(key.isAcceptable()){//在构造函数中注册的key值为OP_ACCEPT,，在判断是否为接收操作
        ServerSocketChannel  ssc = (ServerSocketChannel)key.channel();//获取key值所对应的channel
        SocketChannel sc = ssc.accept();//设置为接收非阻塞通道
        sc.configureBlocking(false);
        sc.register(selector, SelectionKey.OP_READ);//并把这个通道注册为OP_READ      
      }
      if(key.isReadable()){//判断所传的SelectionKey值是否为OP_READ,通过上面的注册后，经过轮询后就会是此操作
        SocketChannel sc = (SocketChannel)key.channel();//获取key对应的channel
        ByteBuffer readbuf = ByteBuffer.allocate(1024);
        int readbytes = sc.read(readbuf);//从channel中读取byte数据并存放readbuf
        if(readbytes > 0){
          readbuf.flip();//检测时候为完整的内容,若不是则返回完整的
          byte[] bytes = new byte[readbuf.remaining()];
          readbuf.get(bytes);
          String string = new String(bytes, "UTF-8");//把读取的数据转换成string
          System.out.println("服务器接受到命令 :"+ string); 
          //"查询时间"就是读取的命令，此字符串要与客户端发送的一致，才能获取当前时间，否则就是bad order
          String currenttime = "查询时间".equalsIgnoreCase(string) ? new java.util.Date(System.currentTimeMillis()).toString() : "bad order";
          dowrite(sc,currenttime);//获取到当前时间后，就需要把当前时间的字符串发送出去
        }else if (readbytes < 0){
          key.cancel();
          sc.close();         
        }else{}       
      }     
    }   
  }
  /**
   * @param sc
   * @param currenttime
   * @throws IOException
   * 服务器的业务操作，将当前时间写到通道内
   */
  private void dowrite(SocketChannel sc, String currenttime) throws IOException {
    System.out.println("服务器 dowrite  currenttime"+  currenttime);
  if(currenttime !=null && currenttime.trim().length()>0){
    byte[] bytes = currenttime.getBytes();//将当前时间序列化
    ByteBuffer writebuf = ByteBuffer.allocate(bytes.length);
    writebuf.put(bytes);//将序列化的内容写入分配的内存
    writebuf.flip();
    sc.write(writebuf); //将此内容写入通道    
  }
 }
}

package bigdata.studynio;
public class NioClient {
  public static void main(String[] args) {
    int port = 8080;
    if(args !=null && args.length > 0){
      try {
        //port = Integer.valueOf(args[0]);
      } catch (Exception e) {
        // TODO: handle exception
      }
    }
    //创建客户端线程
    new Thread(new NioClientWork("127.0.0.1",port),"client").start();
  }
}

package bigdata.studynio;
import java.io.BufferedReader;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class NioClientWork implements Runnable {
  private String host;
  private int port;
  private Selector selector;
  private SocketChannel socketChannel;
  private volatile boolean stop;
  /**
   * @param string
   * @param port
   * 构造函数
   */
  public NioClientWork(String string, int port) {
    this.host = string == null ? "127.0.0.1":string;
    this.port = port;
    try {
      selector= Selector.open();//打开多路复用器
      socketChannel=SocketChannel.open();//打开socketchannel
      socketChannel.configureBlocking(false);
      System.out.println("NIO 客户端启动 端口： "+ port);
    } catch (IOException e) {
      e.printStackTrace();
      System.exit(1);
    }
  }
  /* (non-Javadoc)
   * @see java.lang.Runnable#run()
   */
  @Override
  public void run() {
    try {
      doConnect();//客户端线程需要连接服务器
    } catch (Exception e) {
      e.printStackTrace();
      System.exit(1);
    }
    while(!stop){
      try {
        selector.select(1000);//最大阻塞时间1s
        //获取多路复用器的事件值SelectionKey，并存放在迭代器中
        Set<SelectionKey> selectedKeys = selector.selectedKeys();
        Iterator<SelectionKey> iterator = selectedKeys.iterator();
        SelectionKey key =null;
        while (iterator.hasNext()) {
          key = iterator.next();
          iterator.remove();
          try {
            handleinput(key);//获取多路复用器的事件值SelectionKey，并存放在迭代器中         
          } catch (Exception e) {
            if(key == null){
              key.cancel();
              if(socketChannel ==null)
                socketChannel.close();              
            }
          }         
        }
      } catch (IOException e) {
        e.printStackTrace();
        System.exit(1);
      }     
    }
    if(selector !=null){
      try {
        selector.close();
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  }
  /**
   * @throws IOException
   * 线程连接服务器，并注册操作
   * 
   */
  private void doConnect() throws IOException {
    if(socketChannel.connect(new InetSocketAddress(host, port))){//检测通道是否连接到服务器 
      System.out.println("NIO 客户端 idoConnect OP_READ ");
      socketChannel.register(selector, SelectionKey.OP_READ);//如果已经连接到了服务器，就把通道在selector注册为OP_READ
      dowrite(socketChannel);
    }else{
      System.out.println("NIO 客户端 doConnect OP_CONNECT ");
      socketChannel.register(selector, SelectionKey.OP_CONNECT);//如果客户端未连接到服务器，则将通道注册为OP_CONNECT操作
    } 
  }
  /**
   * @param key
   * @throws IOException
   * 根据SelectionKey的值进行相应的读写操作
   */
  private void handleinput(SelectionKey key) throws IOException {
    //System.out.println("NIO 客户端 handleinput ");
    if(key.isValid()){//判断所传的SelectionKey值是否可用
      //System.out.println("NIO 客户端 isValid() ");
      SocketChannel sc = (SocketChannel) key.channel();
      if(key.isConnectable()){//一开始的时候，客户端需要连接服务器操作，所以检测是否为连接状态
        System.out.println("NIO 客户端 isConnectable ");
        if(sc.finishConnect()){//是否完成连接
          System.out.println("NIO 客户端 finishConnect ");
          dowrite(sc);//向通道内发送数据，就是“查询时间” 的命令,读写通道与通道注册事件类型无关，注册事件只是当有事件来了，就会去处理相应事件
          sc.register(selector, SelectionKey.OP_READ);//如果完成了连接，就把通道注册为 OP_READ操作,用于接收服务器出过来的数据
        }else{
          System.out.println("NIO 客户端 not finishConnect ");
          System.exit(1);       
        }
      }
      if(key.isReadable()){//根据上面注册的selector的通道读事件，进行操作
        System.out.println("NIO 客户端 isReadable() ");
        ByteBuffer readbuf = ByteBuffer.allocate(1024);
        int readbytes = sc.read(readbuf);//获取通道从服务器发过来的数据，并反序列化
        if(readbytes > 0){
          readbuf.flip();
          byte[] bytes=new byte[readbuf.remaining()];
          readbuf.get(bytes);
          String string = new String(bytes, "UTF-8");
          System.out.println("时间是: " + string);
          this.stop=true; //操作完毕后，关闭所有的操作       
        }else if (readbytes < 0){
          key.cancel();
          sc.close();
        }else{}     
       }        
    }   
  }
  private void dowrite(SocketChannel sc) throws IOException {
    byte[] req = "查询时间".getBytes();
    ByteBuffer writebuf = ByteBuffer.allocate(req.length);
    writebuf.put(req);
    writebuf.flip();
    sc.write(writebuf);
    if(!writebuf.hasRemaining()){
      System.out.println("向服务器发送命令成功 ");
    } 
  }
}

/**
 * @author shenj
 * @title: NioBufferOnlyRead
 * @projectName NettyPro
 * @date 2020/12/1 9:54
 */
public class NioBufferOnlyRead {
    public static void main(String[] args) {
        ByteBuffer dataBuffer = ByteBuffer.allocate(5);
        for (int i = 0; i < dataBuffer.capacity() - 1; i++) {
            dataBuffer.put((byte) (i * 2));
        }
        dataBuffer.flip();
        //可以从一个创建的Buffer获取OnlyReadBuffer
        ByteBuffer onlyByteBuffer = dataBuffer.asReadOnlyBuffer();
        while (onlyByteBuffer.hasRemaining()) {
            System.out.println(onlyByteBuffer.get());
        }
        //无法往OnlyReadBuffer写数据
//        onlyByteBuffer.put((byte)(2));
        //可以往dataBuffer中写数据
        dataBuffer.put((byte) (2));
    }
}

/**
 * @author shenj
 * @title: NioBufferAsReadOnlyBuffer
 * @projectName NettyPro
 * @date 2020/12/1 11:35
 */
public class NioBufferAsReadOnlyBuffer {
    public static void main(String[] args) {
        try {
            RandomAccessFile randomAccessFile = new RandomAccessFile("src/main/resources/filedata/data1.txt", "rw");
            FileChannel randomAccessFileChannel = randomAccessFile.getChannel();
            MappedByteBuffer map = randomAccessFileChannel.map(FileChannel.MapMode.READ_WRITE, 0, randomAccessFileChannel.size());
            map.put(0, (byte) ('H'));
            randomAccessFile.close();
            randomAccessFileChannel.close();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

String name = (file != null ? file.getPath() : null);
        int imode = -1;
        if (mode.equals("r"))
            imode = O_RDONLY;
        else if (mode.startsWith("rw")) {
            imode = O_RDWR;
            rw = true;
            if (mode.length() > 2) {
                if (mode.equals("rws"))
                    imode |= O_SYNC;
                else if (mode.equals("rwd"))
                    imode |= O_DSYNC;
                else
                    imode = -1;
            }

public abstract MappedByteBuffer map(MapMode mode,
                                         long position, long size)
        throws IOException;

/**
         * Mode for a read-only mapping.
         */
        public static final MapMode READ_ONLY
            = new MapMode("READ_ONLY");
        /**
         * Mode for a read/write mapping.
         */
        public static final MapMode READ_WRITE
            = new MapMode("READ_WRITE");
        /**
         * Mode for a private (copy-on-write) mapping.
         */
        public static final MapMode PRIVATE
            = new MapMode("PRIVATE");

//实现Closeable接口表示拥有自动关闭流的功能
public abstract class Selector implements Closeable {
//表示获得一个Selector实例
public static Selector open() throws IOException {
//表示设置超时时间，非阻塞,当有事件发生的时候将将注册到相应的SelectionLey中
public abstract int select(long timeout)
//表示获得已经注册了所有的Selectkey
public abstract Set<SelectionKey> selectedKeys();
//阻塞的方法
public abstract int select() throws IOException;
//在未超过select(long timeout)的范围中，可以wakeup唤醒selector
public abstract Selector wakeup();
//不阻塞，立即返回
public abstract int selectNow() throws IOException;

1.当客户端连接时，会通过ServerSocketChannel得到SocketChannel
2.将socketChannel注册到Selector上，register(Selector sel, int ops)一个selector上可以注册多个SocketChannel
3.注册后返回一个SelectionKey，会和该Selector关联（集合）
4.Selector进行监听select方法，返回有事件发生的通道的个数
5.进一步得到各个SelectionKey（有事件发生）
6.在通过SelectionKey反向获取SocketChannel，方法channel（）
7.可以通过得到channel，完成业务处理

/**
 * @author shenj
 * @title: NioServer
 * @projectName NettyPro
 * @date 2020/12/1 17:57
 */
public class NioServer {
    public static void main(String[] args) {
        try {
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            Selector selector = Selector.open();
            serverSocketChannel.socket().bind(new InetSocketAddress(7798));
            //设置未非阻塞
            serverSocketChannel.configureBlocking(false);
            //注册selector，设置关注事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            while (true) {
                if (selector.select(1000) == 0) {
                    System.out.println("超时1s！");
                    continue;
                }
                //selector>0表示触发了关注事件
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                iterator.forEachRemaining(s -> {
                    //监听连接事件
                    if (s.isAcceptable()) {
                        try {
                            SocketChannel socketChannel = serverSocketChannel.accept();
                            //非阻塞
                            socketChannel.configureBlocking(false);
                            System.out.println("一个客户端连接成功"+socketChannel.hashCode());
                            socketChannel.register(selector, OP_READ, ByteBuffer.allocate(1024));
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    //监听读事件
                    if (s.isReadable()) {
                        SocketChannel socketChannel = (SocketChannel)s.channel();
                        ByteBuffer dataBuffer = (ByteBuffer)s.attachment();
                        try {
                            socketChannel.read(dataBuffer);
                            System.out.println("来自客户端:"+new String(dataBuffer.array()));
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        dataBuffer.clear();
                    }
                    iterator.remove();
                });
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

/**
 * @title: NioClient
 * @projectName NettyPro
 * @author shenj
 * @date 2020/12/1 18:57
 */
fun main(args: Array<String>):Unit {
    val data = "我深深地熟悉你脚步的韵律,它在我心中敲击."
    val socketChannel = SocketChannel.open();
    socketChannel.configureBlocking(false);
    val inetSocketAddress = InetSocketAddress("127.0.0.1",7798)
    if(!socketChannel.connect(inetSocketAddress)){
        while (!socketChannel.finishConnect()){
            System.out.println("连接服务器，需要时间....出去溜达会吧~")
        }
    }
    val dataBuffer = ByteBuffer.wrap(data.toByteArray(Charsets.UTF_8))
    socketChannel.write(dataBuffer)
    System.`in`.read()
}

//注册channel对应的SelectionKey
public abstract class SelectionKey {
//得到与之与之关联的selector
public abstract Selector selector();
//得到与之关联的channel
public abstract SelectableChannel channel();
//是否可以读
public final boolean isReadable() 
//是否可以写
public final boolean isWritable() 
//是否可以accept
public final boolean isAcceptable()
//获得与之关联的数据 
public final Object attachment()
//改变监听事件
public abstract SelectionKey interestOps(int ops);

public static final int OP_READ = 1 << 0;//1 读操作
public static final int OP_WRITE = 1 << 2;//4 写操作
public static final int OP_CONNECT = 1 << 3;//8 已经连接
public static final int OP_ACCEPT = 1 << 4;//16 有新的网络可以accept

public abstract class SocketChannel
    extends AbstractSelectableChannel
    implements ByteChannel, ScatteringByteChannel, GatheringByteChannel, NetworkChannel
{
//获得一个SocketChannel通道
public static SocketChannel open() throws IOException {
//设置阻塞或非阻塞
public abstract boolean isConnectionPending();
//连接服务器
public abstract boolean connect(SocketAddress remote) throws IOException;
//如果connect方法连接失败，接下来通过该方法进行完成连接操作
public abstract boolean finishConnect() throws IOException;
//往通道内写数据
public abstract int write(ByteBuffer src) throws IOException;
//往通道里读数据
public abstract int read(ByteBuffer dst) throws IOException;
//注册一个选择器斌设置监听事件。最后一个参数可以设置共享数据
public final SelectionKey register(Selector sel, int ops,
                                       Object att)

/**
 * @author shenj
 * @title: GroupToChatWithServer
 * @projectName NettyPro
 * @date 2020/12/1 22:53
 */
public class GroupToChatWithServer {
    private Selector selector;
    private ServerSocketChannel serverSocketChannel;
    private static final int Port = 7799;
    /**
     * @author shenj
     * @title:  GroupToChatWithServer
     * @date 2020/12/2  14:18
     * 无参构造 
     * */
    public GroupToChatWithServer() {
        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.socket().bind(new InetSocketAddress(Port));
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * @author shenj
     * @title: listen
     * @date 2020/12/2  14:13
     *多路复用器监听channel通道的事件
     */
    public void listen() {
        try {
            while (true) {
                int status = selector.select();
                if (status > 0) {
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> key_Iterator = selectionKeys.iterator();
                    key_Iterator.forEachRemaining(s -> {
                        if (s.isAcceptable()) {
                            try {
                                SocketChannel socketChannel = serverSocketChannel.accept();
                                socketChannel.configureBlocking(false);
                                socketChannel.register(selector, SelectionKey.OP_READ);
                                System.out.println(socketChannel.getRemoteAddress()+"上线了");
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        if(s.isReadable()){
                            readData(s);
                        }
                        key_Iterator.remove();
                    });
                } else {
//                    System.out.println("等待中....");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
        }
    }
    /**
     * @author shenj
     * @title: readData
     * @date 2020/12/2  14:16
     * Listen监听到通道的read事件，读取信息
     */
    private void readData(SelectionKey key){
        SocketChannel socketChannel = null;
        try{
            socketChannel = (SocketChannel) key.channel();
            ByteBuffer dataBuffer = ByteBuffer.allocate(1024);
            int readCount = socketChannel.read(dataBuffer);
            if (readCount>0){
                String msg = new String(dataBuffer.array());
                System.out.println("from 客户端"+msg);
                sendInfo(msg,socketChannel);
            }
        }catch (Exception e){
            try {
                System.out.println(socketChannel.getRemoteAddress()+"下线了");
                key.cancel();
                socketChannel.close();
            } catch (IOException e2) {
                e2.printStackTrace();
            }
        }
    }
    /**
     * @author shenj
     * @title: sendInfo
     * @date 2020/12/2  14:17
     * 发送信息到其他的客户端
     */
    private void sendInfo(String msg, SocketChannel self){
        System.out.println("服务器发送信息中...");
        selector.selectedKeys().stream().forEach(s->{
            Channel targetChannel = s.channel();
            //排除自己发送
            if (targetChannel instanceof SocketChannel && targetChannel != self){
                SocketChannel dest = (SocketChannel)targetChannel;
                ByteBuffer dataBuffer = ByteBuffer.wrap(msg.getBytes());
                try {
                    dest.write(dataBuffer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }
    /**
     * @author shenj
     * @title: main
     * @date 2020/12/2  14:19
     * 主方法
     */
    public static void main(String[] args) {
        GroupToChatWithServer chatWithServer = new GroupToChatWithServer();
        chatWithServer.listen();
    }
}

/**
 * @author shenj
 * @title: GroupClient
 * @projectName NettyPro
 * @date 2020/12/2 0:04
 */
public class GroupClient {
    private final String host = "127.0.0.1";
    private final int port = 7799;
    private Selector selector;
    private SocketChannel socketChannel;
    private String username;
    /**
     * @author shenj
     * @title: GroupClient
     * @date 2020/12/2  14:20 
     * 无参构造
     */
    public GroupClient() {
        try {
            selector = Selector.open();
            socketChannel = SocketChannel.open(new InetSocketAddress(host,port));
            socketChannel.configureBlocking(false);
            socketChannel.register(selector, SelectionKey.OP_READ);
            username = socketChannel.getLocalAddress().toString().substring(1);
            System.out.println(username+"is ok!!!");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * @author shenj
     * @title: sendInfo
     * @date 2020/12/2  14:12
     * 发送信息到服务端
     */
    public  void sendInfo(String info){
        info = username+"说:"+info;
        try {
            socketChannel.write(ByteBuffer.wrap(info.getBytes()));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * @author shenj
     * @title: readInfo
     * @date 2020/12/2  14:11
     * 读取服务器发送的信息
     */
    public void readInfo(){
        try {
            int status = selector.select();
            if (status>0){
                Iterator<SelectionKey> key_iterator = selector.selectedKeys().iterator();
                while (key_iterator.hasNext()) {
                    SelectionKey s = key_iterator.next();
                    if (s.isReadable()) {
                        SocketChannel socketChannel = (SocketChannel) s.channel();
                        ByteBuffer dataBuffer = ByteBuffer.allocate(1024);
                        try {
                            socketChannel.read(dataBuffer);
                            String msg = new String(dataBuffer.array());
                            System.out.println(msg.trim());
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }
                key_iterator.remove();
            }else {
//                System.out.println("没有可用的通道...");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * @author shenj
     * @title: main
     * @date 2020/12/2  14:20
     * 主方法
     */
    public static void main(String[] args) {
        GroupClient chatClient = new GroupClient();
        new Thread(){
            @Override
            public void run() {
                while (true){
                    chatClient.readInfo();
                    try {
                        sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }.start();
        //发送给服务端
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()){
            String str = scanner.nextLine();
            chatClient.sendInfo(str);
        }
    }
}

/**
 * @author shenj
 * @title: NioServer
 * @projectName NettyPro
 * @date 2020/12/2 10:47
 */
public class NioServer implements Serializable {
    private static final String host = "127.0.0.1";
    private static final int port = 8899;
    public static void main(String[] args) throws IOException {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(host, port));
        ByteBuffer dataBuffer = ByteBuffer.allocate(4098);
        while (true) {
            SocketChannel socketChannel = serverSocketChannel.accept();
            int readCount = 0;
            while (-1 != readCount) {
                try {
                    //统计读取字节
                    readCount = socketChannel.read(dataBuffer);
                }catch (IOException e){
                    break;
                }
                //可以让dataBuffer复用，使position = 0;mark = -1作废
                dataBuffer.rewind();
            }
        }
    }
}

/**
 * @author shenj
 * @title: NioClient
 * @projectName NettyPro
 * @date 2020/12/2 14:38
 */
public class NioClient implements Serializable {
    private static final  String fileName = "protoc-3.6.1-win32.zip";
    private static final String  host = "127.0.0.1";
    private static final int port = 8899;
    public static void main(String[] args) throws IOException {
        SocketChannel socketChannel = SocketChannel.open();
        //客户端使用connect连接
        socketChannel.connect(new InetSocketAddress(host,port));
        //文件传输通道
        FileChannel fileChannel = new FileInputStream(new File(fileName)).getChannel();
        long start = System.currentTimeMillis();
        long byteNum = fileChannel.transferTo(0, fileChannel.size(), socketChannel);
        System.out.println("传输的字节数："+byteNum+"耗时："+(System.currentTimeMillis()-start));
        fileChannel.close();
    }
}

package com.anxpp.io.calculator.aio.server;  
/** 
 * AIO服务端 
 * @author shenj 
 * @version 1.0 
 */  
public class Server {  
    private static int DEFAULT_PORT = 12345;  
    private static AsyncServerHandler serverHandle;  
    public volatile static long clientCount = 0;  
    public static void start(){  
        start(DEFAULT_PORT);  
    }  
    public static synchronized void start(int port){  
        if(serverHandle!=null)  
            return;  
        serverHandle = new AsyncServerHandler(port);  
        new Thread(serverHandle,"Server").start();  
    }  
    public static void main(String[] args){  
        Server.start();  
    }  
}

package com.anxpp.io.calculator.aio.server;  
import java.io.IOException;  
import java.net.InetSocketAddress;  
import java.nio.channels.AsynchronousServerSocketChannel;  
import java.util.concurrent.CountDownLatch;  
public class AsyncServerHandler implements Runnable {  
    public CountDownLatch latch;  
    public AsynchronousServerSocketChannel channel;  
    public AsyncServerHandler(int port) {  
        try {  
            //创建服务端通道  
            channel = AsynchronousServerSocketChannel.open();  
            //绑定端口  
            channel.bind(new InetSocketAddress(port));  
            System.out.println("服务器已启动，端口号：" + port);  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }  
    @Override  
    public void run() {  
        //CountDownLatch初始化  
        //它的作用：在完成一组正在执行的操作之前，允许当前的现场一直阻塞  
        //此处，让现场在此阻塞，防止服务端执行完成后退出  
        //也可以使用while(true)+sleep   
        //生成环境就不需要担心这个问题，以为服务端是不会退出的  
        latch = new CountDownLatch(1);  
        //用于接收客户端的连接  
        channel.accept(this,new AcceptHandler());  
        try {  
            latch.await();  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
    }  
}

package com.anxpp.io.calculator.aio.server;  
import java.nio.ByteBuffer;  
import java.nio.channels.AsynchronousSocketChannel;  
import java.nio.channels.CompletionHandler;  
//作为handler接收客户端连接  
public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncServerHandler> {  
    @Override  
    public void completed(AsynchronousSocketChannel channel,AsyncServerHandler serverHandler) {  
        //继续接受其他客户端的请求  
        Server.clientCount++;  
        System.out.println("连接的客户端数：" + Server.clientCount);  
        serverHandler.channel.accept(serverHandler, this);  
        //创建新的Buffer  
        ByteBuffer buffer = ByteBuffer.allocate(1024);  
        //异步读  第三个参数为接收消息回调的业务Handler  
        channel.read(buffer, buffer, new ReadHandler(channel));  
    }  
    @Override  
    public void failed(Throwable exc, AsyncServerHandler serverHandler) {  
        exc.printStackTrace();  
        serverHandler.latch.countDown();  
    }  
}

package com.anxpp.io.calculator.aio.server;  
import java.io.IOException;  
import java.io.UnsupportedEncodingException;  
import java.nio.ByteBuffer;  
import java.nio.channels.AsynchronousSocketChannel;  
import java.nio.channels.CompletionHandler;  
import com.anxpp.io.utils.Calculator;  
public class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {  
    //用于读取半包消息和发送应答  
    private AsynchronousSocketChannel channel;  
    public ReadHandler(AsynchronousSocketChannel channel) {  
            this.channel = channel;  
    }  
    //读取到消息后的处理  
    @Override  
    public void completed(Integer result, ByteBuffer attachment) {  
        //flip操作  
        attachment.flip();  
        //根据  
        byte[] message = new byte[attachment.remaining()];  
        attachment.get(message);  
        try {  
            String expression = new String(message, "UTF-8");  
            System.out.println("服务器收到消息: " + expression);  
            String calrResult = null;  
            try{  
                calrResult = Calculator.cal(expression).toString();  
            }catch(Exception e){  
                calrResult = "计算错误：" + e.getMessage();  
            }  
            //向客户端发送消息  
            doWrite(calrResult);  
        } catch (UnsupportedEncodingException e) {  
            e.printStackTrace();  
        }  
    }  
    //发送消息  
    private void doWrite(String result) {  
        byte[] bytes = result.getBytes();  
        ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);  
        writeBuffer.put(bytes);  
        writeBuffer.flip();  
        //异步写数据 参数与前面的read一样  
        channel.write(writeBuffer, writeBuffer,new CompletionHandler<Integer, ByteBuffer>() {  
            @Override  
            public void completed(Integer result, ByteBuffer buffer) {  
                //如果没有发送完，就继续发送直到完成  
                if (buffer.hasRemaining())  
                    channel.write(buffer, buffer, this);  
                else{  
                    //创建新的Buffer  
                    ByteBuffer readBuffer = ByteBuffer.allocate(1024);  
                    //异步读  第三个参数为接收消息回调的业务Handler  
                    channel.read(readBuffer, readBuffer, new ReadHandler(channel));  
                }  
            }  
            @Override  
            public void failed(Throwable exc, ByteBuffer attachment) {  
                try {  
                    channel.close();  
                } catch (IOException e) {  
                }  
            }  
        });  
    }  
    @Override  
    public void failed(Throwable exc, ByteBuffer attachment) {  
        try {  
            this.channel.close();  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }  
}

package com.anxpp.io.calculator.aio.client;  
import java.util.Scanner;  
public class Client {  
    private static String DEFAULT_HOST = "127.0.0.1";  
    private static int DEFAULT_PORT = 12345;  
    private static AsyncClientHandler clientHandle;  
    public static void start(){  
        start(DEFAULT_HOST,DEFAULT_PORT);  
    }  
    public static synchronized void start(String ip,int port){  
        if(clientHandle!=null)  
            return;  
        clientHandle = new AsyncClientHandler(ip,port);  
        new Thread(clientHandle,"Client").start();  
    }  
    //向服务器发送消息  
    public static boolean sendMsg(String msg) throws Exception{  
        if(msg.equals("q")) return false;  
        clientHandle.sendMsg(msg);  
        return true;  
    }  
    @SuppressWarnings("resource")  
    public static void main(String[] args) throws Exception{  
        Client.start();  
        System.out.println("请输入请求消息：");  
        Scanner scanner = new Scanner(System.in);  
        while(Client.sendMsg(scanner.nextLine()));  
    }  
}

package com.anxpp.io.calculator.aio.client;  
import java.io.IOException;  
import java.net.InetSocketAddress;  
import java.nio.ByteBuffer;  
import java.nio.channels.AsynchronousSocketChannel;  
import java.nio.channels.CompletionHandler;  
import java.util.concurrent.CountDownLatch;  
public class AsyncClientHandler implements CompletionHandler<Void, AsyncClientHandler>, Runnable {  
    private AsynchronousSocketChannel clientChannel;  
    private String host;  
    private int port;  
    private CountDownLatch latch;  
    public AsyncClientHandler(String host, int port) {  
        this.host = host;  
        this.port = port;  
        try {  
            //创建异步的客户端通道  
            clientChannel = AsynchronousSocketChannel.open();  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }  
    @Override  
    public void run() {  
        //创建CountDownLatch等待  
        latch = new CountDownLatch(1);  
        //发起异步连接操作，回调参数就是这个类本身，如果连接成功会回调completed方法  
        clientChannel.connect(new InetSocketAddress(host, port), this, this);  
        try {  
            latch.await();  
        } catch (InterruptedException e1) {  
            e1.printStackTrace();  
        }  
        try {  
            clientChannel.close();  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }  
    //连接服务器成功  
    //意味着TCP三次握手完成  
    @Override  
    public void completed(Void result, AsyncClientHandler attachment) {  
        System.out.println("客户端成功连接到服务器...");  
    }  
    //连接服务器失败  
    @Override  
    public void failed(Throwable exc, AsyncClientHandler attachment) {  
        System.err.println("连接服务器失败...");  
        exc.printStackTrace();  
        try {  
            clientChannel.close();  
            latch.countDown();  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }  
    //向服务器发送消息  
    public void sendMsg(String msg){  
        byte[] req = msg.getBytes();  
        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);  
        writeBuffer.put(req);  
        writeBuffer.flip();  
        //异步写  
        clientChannel.write(writeBuffer, writeBuffer,new WriteHandler(clientChannel, latch));  
    }  
}

package com.anxpp.io.calculator.aio.client;  
import java.io.IOException;  
import java.nio.ByteBuffer;  
import java.nio.channels.AsynchronousSocketChannel;  
import java.nio.channels.CompletionHandler;  
import java.util.concurrent.CountDownLatch;  
public class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {  
    private AsynchronousSocketChannel clientChannel;  
    private CountDownLatch latch;  
    public WriteHandler(AsynchronousSocketChannel clientChannel,CountDownLatch latch) {  
        this.clientChannel = clientChannel;  
        this.latch = latch;  
    }  
    @Override  
    public void completed(Integer result, ByteBuffer buffer) {  
        //完成全部数据的写入  
        if (buffer.hasRemaining()) {  
            clientChannel.write(buffer, buffer, this);  
        }  
        else {  
            //读取数据  
            ByteBuffer readBuffer = ByteBuffer.allocate(1024);  
            clientChannel.read(readBuffer,readBuffer,new ReadHandler(clientChannel, latch));  
        }  
    }  
    @Override  
    public void failed(Throwable exc, ByteBuffer attachment) {  
        System.err.println("数据发送失败...");  
        try {  
            clientChannel.close();  
            latch.countDown();  
        } catch (IOException e) {  
        }  
    }  
}

package com.anxpp.io.calculator.aio.client;  
import java.io.IOException;  
import java.io.UnsupportedEncodingException;  
import java.nio.ByteBuffer;  
import java.nio.channels.AsynchronousSocketChannel;  
import java.nio.channels.CompletionHandler;  
import java.util.concurrent.CountDownLatch;  
public class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {  
    private AsynchronousSocketChannel clientChannel;  
    private CountDownLatch latch;  
    public ReadHandler(AsynchronousSocketChannel clientChannel,CountDownLatch latch) {  
        this.clientChannel = clientChannel;  
        this.latch = latch;  
    }  
    @Override  
    public void completed(Integer result,ByteBuffer buffer) {  
        buffer.flip();  
        byte[] bytes = new byte[buffer.remaining()];  
        buffer.get(bytes);  
        String body;  
        try {  
            body = new String(bytes,"UTF-8");  
            System.out.println("客户端收到结果:"+ body);  
        } catch (UnsupportedEncodingException e) {  
            e.printStackTrace();  
        }  
    }  
    @Override  
    public void failed(Throwable exc,ByteBuffer attachment) {  
        System.err.println("数据读取失败...");  
        try {  
            clientChannel.close();  
            latch.countDown();  
        } catch (IOException e) {  
        }  
    }  
}

package com.anxpp.io.calculator.aio;  
import java.util.Scanner;  
import com.anxpp.io.calculator.aio.client.Client;  
import com.anxpp.io.calculator.aio.server.Server;  
/** 
 * 测试方法 
 * @author shenj
 * @version 1.0 
 */  
public class Test {  
    //测试主方法  
    @SuppressWarnings("resource")  
    public static void main(String[] args) throws Exception{  
        //运行服务器  
        Server.start();  
        //避免客户端先于服务器启动前执行代码  
        Thread.sleep(100);  
        //运行客户端   
        Client.start();  
        System.out.println("请输入请求消息：");  
        Scanner scanner = new Scanner(System.in);  
        while(Client.sendMsg(scanner.nextLine()));  
    }  
}

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

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

1) Bossgroup线程维护 Selector，只关注 Accecpt
2)当接收到 Accept：事件，获取到对应的
Socket Channel，封装成 Nioscoket-并
注册到 Worker线程（事件循环），并进行维护
3)当 Worker线程监听到 selector中通道发生自
己感兴趣的事件后，就进行处理（就由
handler),注意 handler已经加入到通道

1)Nett抽象出两组线程池 Bossgroup专门负责接收客户端的连接，
Workergroup专门负责网络的读写
2) Bossgroup和 Workergroup类型都是 Nioeventloopgroup
3) Nioeventloop Group相当于一个事件循环组，这个组中含有多个事件
循环，每一个事件循环是 Nioeventloop
4) Nioeventloop表示一个不断循环的执行处理任务的线程，每
Nioeventloop都有一个 selector，用于监听绑定在其上的 socket的网
络通讯
5) Nioeventloopgroup可以有多个线程，即可以含有多个 Nioeventloop
6)每个 Boss Nioeventloop 1循环执行的步骤有3步
1.轮询 accept事件
2.处理 accept事件，与 client建立连接，生成 Nioscocket Channel，并将
其注册到某个 worker Nioeventloop上的 selector
3.处理任务队列的任务，即 runalltasks
기每个 Worker Nioeventlod循环执行的步骤
1.轮询read, write事件
2.处理o事件，即read, write事件，在对应 Nioscocket Channel A处理
3.处理任务队列的任务，即 runalltasks
8)每个 Worker Nioeventloop处理业务时，会使用 pipeline（管道
pipeline中包含了 channel，即通过 pipeline可以获取到对应通道，管道
中维护了很多的处理器

/**
 * @author shenj
 * @title: NettyServer
 * @projectName NettyPro
 * @date 2020/12/3 16:27
 */
public class NettyServer {
    public static void main(String[] args) throws Exception {
        //轮询组
        EventLoopGroup bossEventLoop = new NioEventLoopGroup();
        EventLoopGroup workerEventLoop = new NioEventLoopGroup();
        try{
        //创建Server startApplicationConfiguration对象
        ServerBootstrap startApplicationConfiguration = new ServerBootstrap();
        //线程组
        startApplicationConfiguration.group(bossEventLoop, workerEventLoop)
                //通道选择NioServerSocketChannel作为通道的实现
                .channel(NioServerSocketChannel.class)
                //队列连接个数
                .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 NettyHandler());
                    }
                });
        System.out.println("Server is ready....");
        //绑定服务器端口并同步，sync表示同步，且开启服务器，生成channelFuture对象
        ChannelFuture channelFuture = startApplicationConfiguration.bind(6668).sync();
        //对通道关闭进行监听，并且同步
        channelFuture.channel().closeFuture().sync();
        }finally {
            //出现异常，优雅的关闭
            bossEventLoop.shutdownGracefully();
            workerEventLoop.shutdownGracefully();
        }
    }
}

/**
 * @title: NettyServer
 * @projectName NettyPro
 * @author shenj
 * @date 2020/12/3 16:15
 */
fun main() {
    //获得MainReactor和WorkerReactor的无限轮询
    val mainEventGroup: EventLoopGroup = NioEventLoopGroup()
    val workerEventGroup: EventLoopGroup = NioEventLoopGroup()
    try {
        //创建Server StartApplication 对象
        val startApplicationConfiguration = ServerBootstrap()
        //设置startApplication对象的参数
        startApplicationConfiguration.group(mainEventGroup, workerEventGroup)
                .channel(NioServerSocketChannel::class.java)
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childHandler(object : ChannelInitializer<io.netty.channel.socket.SocketChannel>() {
                    override fun initChannel(ch: io.netty.channel.socket.SocketChannel) {
                        ch.pipeline().addLast(NettyHandler())
                    }
                })
        println("Server is ready....")
        val channelFuture: ChannelFuture = startApplicationConfiguration.bind(7788).sync()
//    监听通道关闭事件，同步
        channelFuture.channel().closeFuture().sync()
    } finally {
        mainEventGroup.shutdownGracefully()
        workerEventGroup.shutdownGracefully()
    }
}

/**
 * @author shenj
 * @title: NettyHandler
 * @projectName NettyPro
 * @date 2020/12/3 19:50
 * 注意：要指定自己的Handler就需要继承netty规定好的某个HandlerAdapter（规范）
 * 此时的Handler才能称之为一个Handler
 */
public class NettyHandler extends ChannelInboundHandlerAdapter implements Serializable {
    /**
     * @author shenj
     * @title: channelRead
     * @date 2020/12/3  19:54
     * 1、ChannelHandlerContext:上下文对象，包含pipeline（管道）、channel、地址
     * msg:指的是客户端发送的数据，默认Object对象
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("Server ctx:"+ctx);
        //ByteBuf是netty特有的Buffer性能比NIO的ByteBuffer更好
        ByteBuf dataBuf = (ByteBuf) msg;
        System.out.println("客户端发送来的消息是："+dataBuf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端的地址是："+ctx.channel().remoteAddress());
    }
    /**
     * @author shenj
     * @title: channelReadComplete
     * @date 2020/12/4  1:12
     *  channelRead数据读取完毕的操作
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //writeAndFlush=write+Flush，写入缓冲区并且刷新，使用write只是写到缓冲区
        //非池化的Unpooled，Unpooled.copiedBuffer拷贝到buffer区指定字符集
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello,客户端~",CharsetUtil.UTF_8));
    }
    /**
     * @author shenj
     * @title: exceptionCaught
     * @date 2020/12/4  1:18
     * 处理异常，一般关闭通道
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

/**
 * @title: NettyClient
 * @projectName NettyPro
 * @author shenj
 * @date 2020/12/4 1:38
 */
fun main() {
    val host = "127.0.0.1"
    val port = 6668
    val nioEventLoopGroup:EventLoopGroup = NioEventLoopGroup()
    try {
    //客户端使用Bootstrap
    val startClientApplicationConfiguration = Bootstrap()
    //客户端只需要channel handler
    startClientApplicationConfiguration.group(nioEventLoopGroup)
            .channel(NioSocketChannel::class.java)
            .handler(object : ChannelInitializer<io.netty.channel.socket.SocketChannel>() {
                override fun initChannel(ch: io.netty.channel.socket.SocketChannel?) {
                   ch?.pipeline()?.addLast(NettyClientHandler())
                }
            })
    System.out.println("Client is OK!!!")
    var channelFuture : ChannelFuture = startClientApplicationConfiguration.connect(host, port).sync()
    channelFuture.channel().closeFuture().sync()
    }finally {
        nioEventLoopGroup.shutdownGracefully()
    }
}

/**
 * @title: NettyClientHandler
 * @projectName NettyPro
 * @author shenj
 * @date 2020/12/4 1:52
 *继承自入栈适配器
 */
class NettyClientHandler: ChannelInboundHandlerAdapter() {
    /**
     * @author shenj
     * @title: channelActive
     * @date 2020/12/4  2:04
     * 客户端发送数据给服务器
     */
    override fun channelActive(ctx: ChannelHandlerContext?) {
        println("Client ctx:"+ctx)
        ctx?.writeAndFlush(Unpooled.copiedBuffer("Hi,Server!!!!", CharsetUtil.UTF_8))
    }
    /**
     * @author shenj
     * @title: channelRead
     * @date 2020/12/4  2:04
     * 接收服务器发送的数据并显示
     */
    override fun channelRead(ctx: ChannelHandlerContext?, msg: Any?) {
        val dataBuf = msg as ByteBuf
        println("服务器回复的信息："+dataBuf.toString(CharsetUtil.UTF_8))
        println("服务器地址："+ctx?.channel()?.remoteAddress())
    }
    /**
     * @author shenj
     * @title: exceptionCaught
     * @date 2020/12/4  2:11
     * 异常处理
     */
    override fun exceptionCaught(ctx: ChannelHandlerContext?, cause: Throwable?) {
        cause?.printStackTrace()
        ctx?.close()
    }
}

/**
 * @author shenj
 * @title: ServerHandler
 * @projectName NettyPro
 * @date 2020/12/4 8:18
 */
public class ServerHandler  extends ChannelInboundHandlerAdapter  {
    /**
     * @author shenj
     * @title: channelRead
     * @date 2020/12/4  8:24
     * read
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.channel().eventLoop().execute(()->{
            try {
                Thread.sleep(5*1000);
                ctx.writeAndFlush(Unpooled.copiedBuffer("来自服务器的第二条信息", CharsetUtil.UTF_8));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        ctx.channel().eventLoop().execute(()->{
            try {
                Thread.sleep(5*1000);
                ctx.writeAndFlush(Unpooled.copiedBuffer("来自服务器的第三条信息", CharsetUtil.UTF_8));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        ctx.channel().eventLoop().execute(()->{
            {
                try {
                    Thread.sleep(5*1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("来自服务器的第四条信息", CharsetUtil.UTF_8));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        System.out.println("This is debug");
    }
    /**
     * @author shenj
     * @title: channelReadComplete
     * @date 2020/12/4  8:24
     * channelReadComplete send Info
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("来自服务器的第一条信息",CharsetUtil.UTF_8));
    }
    /**
     * @author shenj
     * @title: exceptionCaught
     * @date 2020/12/4  8:25
     * try Bug
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

/**
 * @author shenj
 * @title: ServerHandler
 * @projectName NettyPro
 * @date 2020/12/4 8:18
 */
public class ServerHandler  extends ChannelInboundHandlerAdapter  {
    /**
     * @author shenj
     * @title: channelRead
     * @date 2020/12/4  8:24
     * read
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.channel().eventLoop().execute(()->{
            try {
                Thread.sleep(5*1000);
                ctx.writeAndFlush(Unpooled.copiedBuffer("来自服务器的第二条信息", CharsetUtil.UTF_8));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        ctx.channel().eventLoop().execute(()->{
            try {
                Thread.sleep(5*1000);
                ctx.writeAndFlush(Unpooled.copiedBuffer("来自服务器的第三条信息", CharsetUtil.UTF_8));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        ctx.channel().eventLoop().schedule(()-> {
                try {
                    Thread.sleep(10*1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("来自服务器的第四条信息", CharsetUtil.UTF_8));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            //TimeUint.SECOND是10的单位（s）
            },10,TimeUnit.SECONDS);
        System.out.println("This is debug");
    }
    /**
     * @author shenj
     * @title: channelReadComplete
     * @date 2020/12/4  8:24
     * channelReadComplete send Info
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("来自服务器的第一条信息",CharsetUtil.UTF_8));
    }
    /**
     * @author shenj
     * @title: exceptionCaught
     * @date 2020/12/4  8:25
     * try Bug
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

ctx.channel().eventLoop().schedule(()-> {
                try {
                    Thread.sleep(10*1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("来自服务器的第四条信息", CharsetUtil.UTF_8));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            },10,TimeUnit.SECONDS);

try {
            ServerBootstrap severStartApplicationConfiguration = new ServerBootstrap();
            severStartApplicationConfiguration.group(mainLoop, workerLoop)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //可在此处命名一个集合用于装每个客户端的SocketChannel,然后通过Handler的ctx.channel().eventLoop().execute/Scheduler
                            //进行用户标识投放广告
                            System.out.println("新加入的客户端HasgCode；"+ch.hashCode());
                            ch.pipeline().addLast(new ServerHandler());
                        }
                    });

Netty的异步模型是建立在 future和 callback的之上的。 callback就是回调。重点说
Future，它的核心思想是：假设一个方法fun，计算过程可能非常耗时，等待fun返回
显然不合适。那么可以在调用fun的时候，立马返回一个 Future，后续可以通过
Future去监控方法fun的处理过程（即： Future-listener机制）

1)表示异步的执行结果，可以通过它提供的方法来检测执行是否完成，比如检索计算等
等
2) Channelfuture是一个接口： public interface Channelfuture extends Future<Void
我们可以添加监听器，当监听的事件发生时，就会通知到监听器。

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

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

ChannelFuture channelFuture = severStartApplicationConfiguration.bind(5566).sync();
//添加监听器，监听事件
channelFuture.addListener((channelFutures) ->{
System.out.println(channelFuture.isSuccess() ? "监听5566端口成功":"监听5566端口失败");
});
channelFuture.channel().closeFuture().sync();

1、HTTP Request 第一部分是包含的头信息
2、HttpContent 里面包含的是数据，可以后续有多个 HttpContent 部分
3、LastHttpContent 标记是 HTTP request 的结束，同时可能包含头的尾部信息
4、完整的 HTTP request，由1，2，3组成

1、HTTP response 第一部分是包含的头信息
2、HttpContent 里面包含的是数据，可以后续有多个 HttpContent 部分
3、LastHttpContent 标记是 HTTP response 的结束，同时可能包含头的尾部信息
4、完整的 HTTP response，由1，2，3组成
从request的介绍我们可以看出来，一次http请求并不是通过一次对话完成的，他中间可能有很次的连接。netty每一次对话都会建立一个channel，并且一个ChannelInboundHandler一般是不会同时去处理多个Channel的。
如何在一个Channel里面处理一次完整的Http请求？这就要用到我们上图提到的FullHttpRequest，我们只需要在使用netty处理channel的时候，只处理消息是FullHttpRequest的Channel，这样我们就能在一个ChannelHandler中处理一个完整的Http请求了。

//通道注册之后
public void channelRegistered(ChannelHandlerContext ctx)
//注销通道
public void channelUnregistered(ChannelHandlerContext ctx)
//通道就绪
public void channelActive(ChannelHandlerContext ctx)
//通道未处于活动
public void channelInactive(ChannelHandlerContext ctx)
//通道读取事件
public void channelRead(ChannelHandlerContext ctx, Object msg)
//通道读取事件之后
public void channelReadComplete(ChannelHandlerContext ctx)
//异常处理
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)

//添加到双链表的首个
ChannelPipeline addFirst(String name, ChannelHandler handler);
//添加到双链表的尾部
ChannelPipeline addLast(String name, ChannelHandler handler);

public NioEventLoopGroup()
//继承自多线程组MultithreadEventExecutorGroup的线程关闭
public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit)

for (int i = 0; i <dataBuf.capacity() ; i++) {
      System.out.println(dataBuf.readByte());
        }

for (int i = 0; i < dataBuf.capacity(); i++) {
      dataBuf.writeByte(i);
        }

//获取ByteBuf的容量大小
public abstract int capacity();
//在原有Buf基础上设置新的容量，并返回一个Buf，注意：原来的Buf也将变化
public abstract ByteBuf capacity(int newCapacity);
//最大可设置 
public abstract int maxCapacity();
//read的索引
public abstract int readerIndex();
//在原有Buf基础上设置readIndex，并返回一个Buf，注意：原来的Buf也将变化
public abstract ByteBuf readerIndex(int readerIndex);
//在原有Buf基础上设置writeIndex，并返回一个Buf，注意：原来的Buf也将变化
public abstract ByteBuf writerIndex(int writerIndex);
//读取Buf中从指定位置，读取指定长度且指定字符集的字符序列String
public abstract CharSequence getCharSequence(int index, int length, Charset charset);