1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
/**
 * @author ：zsy
 * @date ：Created 2021/11/23 19:55
 * @description：
 */
@Slf4j
public class AioFileChannel {

    public static void main(String[] args) {
        try (AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get("data.txt"), StandardOpenOption.READ)) {

            ByteBuffer buffer = ByteBuffer.allocate(16);
            log.debug("read begin...");
            // 参数1：ByteBuffer
            // 参数2：读取的起始位置
            // 参数3：附加
            // 参数4：回调函数，以守护线程的形式回调
            channel.read(buffer, 0, buffer, new CompletionHandler<Integer, ByteBuffer>() {
                @Override
                public void completed(Integer result, ByteBuffer attachment) {
                    log.debug("read completed...{}", result);
                    attachment.flip();
                    // System.out.println(Charset.defaultCharset().decode(attachment));
                    debugRead(buffer);
                }

                @Override
                public void failed(Throwable exc, ByteBuffer attachment) {
                    exc.printStackTrace();
                }
            });
            log.debug("read end");
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
public class AioServer {
    public static void main(String[] args) throws IOException {
        AsynchronousServerSocketChannel ssc = AsynchronousServerSocketChannel.open();
        ssc.bind(new InetSocketAddress(8080));
        ssc.accept(null, new AcceptHandler(ssc));
        System.in.read();
    }

    private static void closeChannel(AsynchronousSocketChannel sc) {
        try {
            System.out.printf("[%s] %s closen", Thread.currentThread().getName(), sc.getRemoteAddress());
            sc.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {
        private final AsynchronousSocketChannel sc;

        public ReadHandler(AsynchronousSocketChannel sc) {
            this.sc = sc;
        }

        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            try {
                if (result == -1) {
                    closeChannel(sc);
                    return;
                }
                System.out.printf("[%s] %s readn", Thread.currentThread().getName(), sc.getRemoteAddress());
                attachment.flip();
                System.out.println(Charset.defaultCharset().decode(attachment));
                attachment.clear();
                // 处理完第一个 read 时，需要再次调用 read 方法来处理下一个 read 事件
                sc.read(attachment, attachment, this);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            closeChannel(sc);
            exc.printStackTrace();
        }
    }

    private static class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {
        private final AsynchronousSocketChannel sc;

        private WriteHandler(AsynchronousSocketChannel sc) {
            this.sc = sc;
        }

        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            // 如果作为附件的 buffer 还有内容，需要再次 write 写出剩余内容
            if (attachment.hasRemaining()) {
                sc.write(attachment);
            }
        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            exc.printStackTrace();
            closeChannel(sc);
        }
    }

    private static class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> {
        private final AsynchronousServerSocketChannel ssc;

        public AcceptHandler(AsynchronousServerSocketChannel ssc) {
            this.ssc = ssc;
        }

        @Override
        public void completed(AsynchronousSocketChannel sc, Object attachment) {
            try {
                System.out.printf("[%s] %s connectedn", Thread.currentThread().getName(), sc.getRemoteAddress());
            } catch (IOException e) {
                e.printStackTrace();
            }
            ByteBuffer buffer = ByteBuffer.allocate(16);
            // 读事件由 ReadHandler 处理
            sc.read(buffer, buffer, new ReadHandler(sc));
            // 写事件由 WriteHandler 处理
            sc.write(Charset.defaultCharset().encode("server hello!"), ByteBuffer.allocate(16), new WriteHandler(sc));
            // 处理完第一个 accpet 时，需要再次调用 accept 方法来处理下一个 accept 事件
            ssc.accept(null, this);
        }

        @Override
        public void failed(Throwable exc, Object attachment) {
            exc.printStackTrace();
        }
    }
}

1
2
3
4
5
6
7
8
9
File f = new Flie("helloworld/data.txt");
RandomAccessFile flie = RandomAccessFile(f, "r");

byte[] buf = new byte[(int) f.length];
file.read(buf);

Socket socket = ...;
socket.getOutputStream().write(buf);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
/**
 * @author ：zsy
 * @date ：Created 2021/11/19 20:33
 * @description：测试ByteBuffer
 */
@Slf4j
public class TestByteBuffer {
    public static void main(String[] args) {
        // FileChannel
        // 通过输入输出流获取文件 或者 RandomAccessFile
        // try-with-resource语法无需自己写代码关闭资源，资源必须实现AutoClosable接口，重写close方法
        //原理：编译器自动帮我们生成了finally块，并且在里面调用了资源的close方法
        try (FileChannel channel = new FileInputStream("data.txt").getChannel()) {
            // 准备缓冲区
            ByteBuffer buffer = ByteBuffer.allocate(10);
            while (true) {
                // 从channel中读数据，向buffer写入数据
                int len = channel.read(buffer);
                log.debug("读到字节数：{}", len);
                if (len == -1) break;
                // 切换至读模式
                buffer.flip();
                // 打印buffer中的内容
                while (buffer.hasRemaining()) { // 还有剩余未读数据
                    byte b = buffer.get();
                    System.out.print((char) b);
                }
                // 切换至写模式
                buffer.clear();
            }
        } catch (IOException e) {
        }
        ;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
public class Demo {
  public static void main(String[] args) {
    BufferedInputStream bin = null;
    BufferedOutputStream bout = null;
    try {
      bin = new BufferedInputStream(new FileInputStream(new File("test.txt")));
      bout = new BufferedOutputStream(new FileOutputStream(new File("out.txt")));
      int b;
      while ((b = bin.read()) != -1) {
        bout.write(b);
      }
    }
    catch (IOException e) {
      e.printStackTrace();
    }
    finally {
      if (bin != null) {
        try {
          bin.close();
        }
        catch (IOException e) {
          throw e;
        }
        finally {
          if (bout != null) {
            try {
              bout.close();
            }
            catch (IOException e) {
              throw e;
            }
          }
        }
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
/**
 * @author ：zsy
 * @date ：Created 2021/11/19 20:51
 * @description：
 */
@Slf4j
public class TestByteBuffer0 {

    public static void main(String[] args) {

        try (RandomAccessFile file = new RandomAccessFile("data.txt", "rw")) {
            FileChannel channel = file.getChannel();
            ByteBuffer buffer = ByteBuffer.allocate(10);
            while (true) {
                int len = channel.read(buffer);
                buffer.flip();
                log.debug("读取到的字节数：{}", len);
                if (len == -1) break;
                while (buffer.hasRemaining()) {
                    byte b = buffer.get();
                    System.out.println((char) b);
                }
                buffer.clear();
            }
        } catch (IOException e) {
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
/**
 * @author ：zsy
 * @date ：Created 2021/11/20 11:48
 * @description：
 */
public class TestByteBufferAllocate {
    public static void main(String[] args) {
        System.out.println(ByteBuffer.allocate(16));            // java.nio.HeapByteBuffer[pos=0 lim=16 cap=16]
        System.out.println(ByteBuffer.allocateDirect(16));      // java.nio.DirectByteBuffer[pos=0 lim=16 cap=16]
    }
}

1
2
int readBytes = channel.read(buf);

1
2
buf.put((byte)127);

1
2
int writeBytes = channel.write(buf);

1
2
byte b = buf.get();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
/**
 * @author ：zsy
 * @date ：Created 2021/11/20 12:07
 * @description：
 */
public class TestByteBufferRead {

    public static void main(String[] args) {
        ByteBuffer buffer = ByteBuffer.allocate(10);
        buffer.put(new byte[]{'a', 'b', 'c', 'd'});
        buffer.flip();

        /*buffer.get(new byte[4]);
        buffer.rewind();
        debugAll(buffer);*/

        System.out.println((char) buffer.get());
        System.out.println((char) buffer.get());
        buffer.mark();
        System.out.println((char) buffer.get());
        System.out.println((char) buffer.get());
        debugAll(buffer);
        buffer.reset();
        debugAll(buffer);

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
 * @author ：zsy
 * @date ：Created 2021/11/20 12:25
 * @description：
 */
public class TestByteBufferString {

    public static void main(String[] args) {
        ByteBuffer buffer = ByteBuffer.allocate(16);
        buffer.put("hello Netty".getBytes());
        debugAll(buffer);

        ByteBuffer buffer2 = Charset.forName("utf-8").encode("您好");
        debugAll(buffer2);

        CharBuffer buffer3 = StandardCharsets.UTF_8.decode(buffer2);
        System.out.println(buffer3.toString());
        
        ByteBuffer buffer4 = ByteBuffer.wrap("hello".getBytes());
        debugAll(buffer4);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
 * @author ：zsy
 * @date ：Created 2021/11/20 15:21
 * @description：
 */
public class TestScatteringReads {

    public static void main(String[] args) {
        try (FileChannel channel = new FileInputStream("word1.txt").getChannel()) {
            ByteBuffer a = ByteBuffer.allocate(3);
            ByteBuffer b = ByteBuffer.allocate(3);
            ByteBuffer c = ByteBuffer.allocate(5);
            channel.read(new ByteBuffer[]{a, b, c});
            a.flip();
            b.flip();
            c.flip();
            debugAll(a);
            debugAll(b);
            debugAll(c);
        } catch (IOException e) {
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
 * @author ：zsy
 * @date ：Created 2021/11/20 15:32
 * @description：
 */
public class TestGatherWrites {

    public static void main(String[] args) {
        try (RandomAccessFile randomAccessFile = new RandomAccessFile("word2.txt", "rw")) {
            FileChannel channel = randomAccessFile.getChannel();
            final ByteBuffer b1 = StandardCharsets.UTF_8.encode("Hello");
            final ByteBuffer b2 = StandardCharsets.UTF_8.encode("Netty");
            final ByteBuffer b3 = StandardCharsets.UTF_8.encode("您好");
            channel.write(new ByteBuffer[]{b1, b2, b3});
        } catch (IOException e) {
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
/**
 * @author ：zsy
 * @date ：Created 2021/11/20 15:54
 * @description：
 */
public class TestByteBufferExam {

    public static void main(String[] args) {
        ByteBuffer source = ByteBuffer.allocate(32);
        source.clear();
        source.put("Hello WorldnI'm zhangsannHo".getBytes());
        split(source);
        source.put("w are youn".getBytes());
        split(source);
    }

    private static void split(ByteBuffer source) {
        source.flip();
        for (int i = 0; i < source.limit(); i++) {
            if (source.get(i) == 'n') {
                int len = i + 1 - source.position();
                ByteBuffer tmp = ByteBuffer.allocate(len);
                tmp.clear();
                for (int j = 0; j < len; j++) {
                    tmp.put(source.get());
                }
                tmp.flip();
                debugAll(tmp);
            }
        }
        //可能没有读完，不能使用clear
        source.compact();
    }
}


+--------+-------------------- all ------------------------+----------------+
position: [0], limit: [12]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 20 57 6f 72 6c 64 0a             |Hello World.    |
+--------+-------------------------------------------------+----------------+
+--------+-------------------- all ------------------------+----------------+
position: [0], limit: [13]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 49 27 6d 20 7a 68 61 6e 67 73 61 6e 0a          |I'm zhangsan.   |
+--------+-------------------------------------------------+----------------+
+--------+-------------------- all ------------------------+----------------+
position: [0], limit: [12]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 6f 77 20 61 72 65 20 79 6f 75 0a             |How are you.    |
+--------+-------------------------------------------------+----------------+

1
2
int readBytes = channel.read(buffer);

1
2
3
4
5
6
7
8
ByteBuffer buffer = ...;
buffer.put(...); // 存入数据
buffer.flip();   // 切换读模式

while(buffer.hasRemaining()) {
    channel.write(buffer);
}

1
2
long pos = channel.position();

1
2
3
long newPos = ...;
channel.position(newPos);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
 * @author ：zsy
 * @date ：Created 2021/11/20 16:33
 * @description：
 */
public class TestFileChannelTransferTo {

    public static void main(String[] args) {
        try (
                FileChannel from = new FileInputStream("data.txt").getChannel();
                FileChannel to = new RandomAccessFile("to.txt", "rw").getChannel()
        ) {
            long size = from.size();
            for (long left = size; left > 0;) {
                System.out.println("position：" + (size - left) + "   count：" + left);
                left -= from.transferTo(left - size, left, to);
            }
        } catch (IOException e) {
            e.printStackTrace();;
        }
    }

}

1
2
3
4
5
6
7
8
Path source = Paths.get("1.txt"); // 相对路径 使用 user.dir 环境变量来定位 1.txt

Path source = Paths.get("d:\1.txt"); // 绝对路径 代表了  d:1.txt

Path source = Paths.get("d:/1.txt"); // 绝对路径 同样代表了  d:1.txt

Path projects = Paths.get("d:\data", "projects"); // 代表了  d:dataprojects

1
2
3
4
5
6
d:
	|- data
		|- projects
			|- a
			|- b

1
2
3
4
Path path = Paths.get("d:\data\projects\a\..\b");
System.out.println(path);
System.out.println(path.normalize()); // 正常化路径

1
2
3
d:dataprojectsa..b
d:dataprojectsb

1
2
3
Path path = Paths.get("helloword/data.txt");
System.out.println(Files.exists(path));

1
2
3
Path path = Paths.get("helloword/d1");
Files.createDirectory(path);

1
2
3
Path path = Paths.get("helloword/d1/d2");
Files.createDirectories(path);

1
2
3
4
5
Path source = Paths.get("helloword/data.txt");
Path target = Paths.get("helloword/target.txt");

Files.copy(source, target);

1
2
Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING);

1
2
3
4
5
Path source = Paths.get("helloword/data.txt");
Path target = Paths.get("helloword/data.txt");

Files.move(source, target, StandardCopyOption.ATOMIC_MOVE);

1
2
3
4
Path target = Paths.get("helloword/target.txt");

Files.delete(target);

1
2
3
4
Path target = Paths.get("helloword/d1");

Files.delete(target);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
/**
 * @author ：zsy
 * @date ：Created 2021/11/20 22:27
 * @description：
 */
public class TestFilesWalkFileTree {
    public static void main(String[] args) throws IOException {
        AtomicInteger dirCount = new AtomicInteger();
        AtomicInteger fileCount = new AtomicInteger();
        AtomicInteger jarCount = new AtomicInteger();
        Files.walkFileTree(Paths.get("E:\java tools\JDK8\jdk1.8.0_101"), new SimpleFileVisitor<Path>(){
            @Override
            public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
                System.out.println("====>" + dir);
                dirCount.getAndIncrement();
                return super.preVisitDirectory(dir, attrs);
            }

            @Override
            public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
                System.out.println(file);
                fileCount.getAndIncrement();
                if (file.toFile().getName().endsWith(".jar")) {
                    jarCount.getAndIncrement();
                }
                return super.visitFile(file, attrs);
            }
        });
        System.out.println("目录数量：" + dirCount);
        System.out.println("文件数量：" + fileCount);
        System.out.println("jar文件数量：" + jarCount);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public void remove_dir() throws IOException {
    Files.walkFileTree(Paths.get("C:\Users\dell\Desktop\target"), new SimpleFileVisitor<Path>() {
        @Override
        public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
            System.out.println(file.toFile().toString());
            Files.delete(file);
            return super.visitFile(file, attrs);
        }
        @Override
        public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
            System.out.println("====>" + dir);
            return super.preVisitDirectory(dir, attrs);
        }
        @Override
        public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
            System.out.println("删除" + dir);
            Files.delete(dir);
            return super.postVisitDirectory(dir, exc);
        }
    });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
/**
 * @author ：zsy
 * @date ：Created 2021/11/21 10:24
 * @description：
 */
public class TestFileCopy {

    public static void main(String[] args) throws IOException {
        String source = "E:\java tools\JDK8\jdk1.8.0_101";
        String target = "E:\java tools\JDK8\jdk1.8.0_101_new";
        Files.walk(Paths.get(source)).forEach(path -> {
            try {
                String targetName = path.toString().replace(source, target);
                if (Files.isDirectory(path)) {
                    Files.createDirectory(Paths.get(targetName));
                } else if (Files.isRegularFile(path)) {
                    Files.copy(path, Paths.get(targetName));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@Slf4j
public class Server {

    public static void main(String[] args) throws IOException {
        // 创建缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(16);

        // 创建服务器
        ServerSocketChannel ssc = ServerSocketChannel.open();

        // 绑定端口号
        ssc.bind(new InetSocketAddress(8080));

        // 创建连接集合
        List<SocketChannel> channels = new ArrayList<>();

        // 循环接受多个客户端发送的数据
        while (true) {

            // accept用来和客户端建立连接，SocketChannel用来与客户端通信
            log.debug("connecting...");
            SocketChannel curChannel = ssc.accept(); // 阻塞方法，线程停止运行
            log.debug("connected->{}", curChannel);
            channels.add(curChannel);

            // 循环读取每个channel中的数据
            for (SocketChannel channel : channels) {
                log.debug("before read->{}", channel);
                channel.read(buffer); // 阻塞方法，线程停止运行
                buffer.flip();
                debugRead(buffer);
                buffer.clear();
                log.debug("after read->{}", channel);
            }
        }
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
 * @author ：zsy
 * @date ：Created 2021/11/21 11:18
 * @description：
 */
public class Client {

    public static void main(String[] args) throws IOException {
        // 创建客户端
        SocketChannel sc = SocketChannel.open();

        // 连接服务端
        sc.connect(new InetSocketAddress("localhost", 8080));

        System.out.println("wait......");
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
/**
 * @author ：zsy
 * @date ：Created 2021/11/21 11:03
 * @description： 阻塞模式（单线程）
 */
@Slf4j
public class Server {

    public static void main(String[] args) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(16);
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.bind(new InetSocketAddress(8080));
        // 设置为非阻塞模式，此时accept()方法将不会阻塞
        ssc.configureBlocking(false);
        List<SocketChannel> channels = new ArrayList<>();
        while (true) {
            // log.debug("connecting...");
            SocketChannel curChannel = ssc.accept();
            if (curChannel != null) {
                log.debug("connect->{}", curChannel);
                // 设置为非阻塞状态，此时read()方法将不会阻塞
                curChannel.configureBlocking(false);
                channels.add(curChannel);
            }
            for (SocketChannel channel : channels) {
                int read = channel.read(buffer);
                if (read > 0) {
                    buffer.flip();
                    debugRead(buffer);
                    buffer.clear();
                    log.debug("after read...{}", channel);
                }
            }
        }

    }

}

1
2
Selector selector = Selector.open();

1
2
3
channel.configureBlocking(false);
SelectionKey key = channel.register(selector, 绑定事件);

1
2
int count = selector.select();

1
2
int count = selector.select(long timeout);

1
2
int count = selector.selectNow();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
 * @author ：zsy
 * @date ：Created 2021/11/21 16:15
 * @description：
 */
public class Client {

    public static void main(String[] args) {
        try (SocketChannel sc = SocketChannel.open()) {
            sc.connect(new InetSocketAddress("localhost", 8080));

            System.out.println("wait....");
        } catch (IOException e) {
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
/**
 * @author ：zsy
 * @date ：Created 2021/11/21 16:00
 * @description：
 */
@Slf4j
public class Server {

    public static void main(String[] args) throws IOException {
        // 创建Selector
        Selector selector = Selector.open();

        // 创建服务端通道，设置为非阻塞
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);

        // 将服务端通道注册到Selector中
        // SelectionKey就是事件发生后，通过它可以知道事件和那个channel的事件(客户端channel)
        SelectionKey sscKey = ssc.register(selector, 0, null);

        // 设置关注事件
        sscKey.interestOps(SelectionKey.OP_ACCEPT); // 四种事件：accept、connect、read、write

        log.debug("register key -> {}", sscKey);

        // 绑定端口
        ssc.bind(new InetSocketAddress(8080));

        while (true) {
            // select()方法没有事件发生，线程阻塞，有事件发生，线程恢复运行
            // 在事件未处理是不会阻塞，事件发生后要么处理，要么取消
            selector.select();

            // 从selector中获取事件发生的所有SelectionKeys
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();

            // 通过迭代器遍历
            while (iterator.hasNext()) {
                SelectionKey sKey = iterator.next();
                log.debug("sKey -> {}", sKey);
                // 手动删除selectedKeys中的当前selectionKey
                iterator.remove();

                if (sKey.isAcceptable()) { // 如果是一个accept事件
                    // 获取对应channel
                    ServerSocketChannel channel = (ServerSocketChannel) sKey.channel();
                    // log.debug("serverSocketChannel -> {}", channel);
                    SocketChannel sc = channel.accept();
                    log.debug("socketChannel -> {}", sc);

                    // 注册当前socketChannel
                    sc.configureBlocking(false);
                    SelectionKey scKey = sc.register(selector, 0, null);
                    scKey.interestOps(SelectionKey.OP_READ);
                    log.debug("scKey -> {}", scKey);
                } else if (sKey.isReadable()) { // 如果是一个读事件
                    try {
                        SocketChannel sc = (SocketChannel) sKey.channel();
                        // 创建缓冲区
        				ByteBuffer buffer = ByteBuffer.allocate(16);
                        int read = sc.read(buffer);
                        if (read > 0) {
                            buffer.flip();
                            debugAll(buffer);
                            buffer.clear();
                        } else {
                            // 因为关闭连接也是一个读时间
                            // 如果客户端主动关闭，需要手动取消事件
                            sKey.cancel();
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                        // 如果客户端强制关闭，则会触发异常，需要手动取消事件
                        sKey.cancel();
                    }
                }
            }
        }

    }

}

1
2
sc.write(Charset.defaultCharset().encode("Hello"));

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
/**
 * @author ：zsy
 * @date ：Created 2021/11/21 16:00
 * @description：
 */
@Slf4j
public class Server {

    public static void main(String[] args) throws IOException {
        // 创建Selector
        Selector selector = Selector.open();

        // 创建服务端通道，设置为非阻塞
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);

        // 将服务端通道注册到Selector中
        // SelectionKey就是事件发生后，通过它可以知道事件和那个channel的事件(客户端channel)
        SelectionKey sscKey = ssc.register(selector, 0, null);

        // 设置关注事件
        sscKey.interestOps(SelectionKey.OP_ACCEPT); // 四种事件：accept、connect、read、write

        log.debug("register key -> {}", sscKey);

        // 绑定端口
        ssc.bind(new InetSocketAddress(8080));

        while (true) {
            // select()方法没有事件发生，线程阻塞，有事件发生，线程恢复运行
            // 在事件未处理是不会阻塞，事件发生后要么处理，要么取消
            selector.select();

            // 从selector中获取事件发生的所有SelectionKeys
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();

            // 通过迭代器遍历
            while (iterator.hasNext()) {
                SelectionKey sKey = iterator.next();
                log.debug("sKey -> {}", sKey);
                // 手动删除selectedKeys中的当前selectionKey
                iterator.remove();

                if (sKey.isAcceptable()) { // 如果是一个accept事件
                    // 获取对应channel
                    ServerSocketChannel channel = (ServerSocketChannel) sKey.channel();
                    // log.debug("serverSocketChannel -> {}", channel);
                    SocketChannel sc = channel.accept();
                    log.debug("socketChannel -> {}", sc);

                    // 注册当前socketChannel
                    sc.configureBlocking(false);
                    ByteBuffer buffer = ByteBuffer.allocate(16);

                    // attachment 附件，绑定当前selectKey
                    SelectionKey scKey = sc.register(selector, 0, buffer);
                    scKey.interestOps(SelectionKey.OP_READ);
                    log.debug("scKey -> {}", scKey);
                } else if (sKey.isReadable()) { // 如果是一个读事件
                    try {
                        SocketChannel sc = (SocketChannel) sKey.channel();
                        ByteBuffer buffer = (ByteBuffer) sKey.attachment();
                        int read = sc.read(buffer);
                        if (read > 0) {
                            split(buffer);
                            if (buffer.position() == buffer.limit()) {
                                ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
                                buffer.flip();
                                newBuffer.put(buffer);
                                sKey.attach(newBuffer);
                            }
                        } else {
                            // 因为关闭连接也是一个读时间
                            // 如果客户端主动关闭，需要手动取消事件
                            sKey.cancel();
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                        // 如果客户端强制关闭，则会触发异常，需要手动取消事件
                        sKey.cancel();
                    }
                }
            }
        }

    }

    private static void split(ByteBuffer buffer) {
        buffer.flip();
        for (int i = 0; i < buffer.limit(); i++) {
            if (buffer.get(i) == 'n') {
                int len = i + 1 - buffer.position();
                ByteBuffer tmp = ByteBuffer.allocate(len);
                for (int j = 0; j < len; j++) {
                    tmp.put(buffer.get());
                }
                tmp.flip();
                debugRead(tmp);
            }
        }
        buffer.compact();
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
 * @author ：zsy
 * @date ：Created 2021/11/21 16:15
 * @description：
 */
public class Client {

    public static void main(String[] args) {
        try (SocketChannel sc = SocketChannel.open()) {
            sc.connect(new InetSocketAddress("localhost", 8080));
            sc.write(Charset.defaultCharset().encode("HellonWorld-Netty-Zhang-Sann"));
            System.in.read();
        } catch (IOException e) {
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
/**
 * @author ：zsy
 * @date ：Created 2021/11/22 0:14
 * @description：
 */
public class WriteServer {
    public static void main(String[] args) throws IOException {
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        ssc.bind(new InetSocketAddress(8080));

        Selector selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT, null);

        while (true) {
            selector.select();

            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();

            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();

                if (key.isAcceptable()) {
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    SelectionKey scKey = sc.register(selector, SelectionKey.OP_READ, null);

                    // 向客户端发送数据
                    StringBuilder builder = new StringBuilder();
                    for (int i = 0; i < 30000000; i++) {
                        builder.append("a");
                    }
                    ByteBuffer buffer = Charset.defaultCharset().encode(builder.toString());
                    int write = sc.write(buffer);
                    System.out.println("实际写了多少字节：" + write);

                    // 如果剩余字节没有写完，需要注册事件
                    if (buffer.hasRemaining()) {
                        // 在原有读事件的基础上注册写事件
                        scKey.interestOps(scKey.interestOps() + SelectionKey.OP_WRITE);

                        // 将buffer作为附件加到scKey中
                        scKey.attach(buffer);
                    }
                } else if(key.isWritable()) {
                    ByteBuffer buffer = (ByteBuffer) key.attachment();
                    SocketChannel sc = (SocketChannel) key.channel();
                    int write = sc.write(buffer);
                    System.out.println("实际写了多少字节：" + write);
                    if (!buffer.hasRemaining()) {
                        key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);
                        // 删除缓冲区
                        key.attach(null);
                    }
                }
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
/**
 * @author ：zsy
 * @date ：Created 2021/11/22 0:20
 * @description：
 */
public class WriteClient {
    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.configureBlocking(false);

        Selector selector = Selector.open();
        sc.register(selector, SelectionKey.OP_CONNECT + SelectionKey.OP_READ, null);
        sc.connect(new InetSocketAddress("localhost", 8080));

        int count = 0;
        while (true) {
            selector.select();

            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                if (key.isConnectable()) {
                    System.out.println(sc.finishConnect());
                } else if (key.isReadable()) {
                    ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
                    count += sc.read(buffer);
                    buffer.clear();
                    System.out.println(count);
                }

            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
/**
 * @author ：zsy
 * @date ：Created 2021/11/22 13:47
 * @description：
 */
public class WriteServer {

    public static void main(String[] args) throws IOException {
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        ssc.bind(new InetSocketAddress(8080));

        Selector select = Selector.open();
        ssc.register(select, SelectionKey.OP_ACCEPT, null);

        while (true) {
            select.select();
            Iterator<SelectionKey> iterator = select.selectedKeys().iterator();

            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                if (key.isAcceptable()) {
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    Map<String, ByteBuffer> map = new HashMap<>();
                    map.put("read", ByteBuffer.allocate(4));
                    SelectionKey scKey = sc.register(select, SelectionKey.OP_READ, map);

                    StringBuilder builder = new StringBuilder();
                    for (int i = 0; i < 30000000; i++) {
                        builder.append("a");
                    }
                    ByteBuffer buffer = Charset.defaultCharset().encode(builder.toString());

                    int write = sc.write(buffer);
                    System.out.println("实际写了多少字节：" + write);

                    if (buffer.hasRemaining()) {
                        scKey.interestOps(scKey.interestOps() + SelectionKey.OP_WRITE);
                        map.put("write", buffer);
                    }
                } else if (key.isWritable()) {
                    ByteBuffer buffer = (ByteBuffer) ((HashMap) key.attachment()).get("write");
                    SocketChannel sc = (SocketChannel) key.channel();
                    int write = sc.write(buffer);
                    System.out.println("实际写了字节数：" + write);
                    if (!buffer.hasRemaining()) {
                        key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);
                        key.attach(null);
                    }
                } else if (key.isReadable()) {
                    try {
                        SocketChannel sc = (SocketChannel) key.channel();
                        Map<String, ByteBuffer> map = (HashMap) key.attachment();
                        ByteBuffer buffer = map.get("read");
                        int read = sc.read(buffer);
                        if (read > 0) {
                            split(buffer);
                            if (buffer.position() == buffer.limit()) {
                                ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
                                // 切换buffer至读模式
                                buffer.flip();
                                newBuffer.put(buffer);
                                map.put("read", newBuffer);
                            }
                        } else {
                            key.cancel();
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                        key.cancel();
                    }
                }
            }
        }
    }

    private static void split(ByteBuffer buffer) {
        buffer.flip();
        for (int i = 0; i < buffer.limit(); i++) {
            char c = (char) buffer.get(i);
            if (buffer.get(i) == 'n') {
                int len = i + 1 - buffer.position();
                ByteBuffer tmp = ByteBuffer.allocate(len);
                for (int j = 0; j < len; j++) {
                    tmp.put(buffer.get());
                }
                tmp.flip();
                System.out.println("客户端发送的数据 -> " + Charset.defaultCharset().decode(tmp));
            }
        }
        buffer.compact();
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
/**
 * @author ：zsy
 * @date ：Created 2021/11/22 14:00
 * @description：
 */
public class WriteClient {

    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.configureBlocking(false);
        sc.connect(new InetSocketAddress("localhost", 8080));

        Selector selector = Selector.open();
        sc.register(selector, SelectionKey.OP_CONNECT, null);

        int count = 0;
        while (true) {
            selector.select();

            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();

            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();

                if (key.isConnectable()) {
                    if (sc.finishConnect()) {
                        key.interestOps(key.interestOps() + SelectionKey.OP_READ);
                        sc.write(Charset.defaultCharset().encode("HellonWorld-Netty-Zhang-Sann"));
                    } else {
                        key.cancel();
                    }
                } else if (key.isReadable()) {
                    ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
                    count += sc.read(buffer);
                    System.out.println(count);
                    if (!buffer.hasRemaining()) {
                        key.interestOps(key.interestOps() - SelectionKey.OP_READ);
                    }
                }
            }
        }
    }

}