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
    @Test
    public void fileChannelRead() {

        try {
            // 开启FileChannel
            RandomAccessFile aFile = new RandomAccessFile("D:\job\git\java-study\nio\src\main\resources\fileChanel.txt", "rw");
            FileChannel inChannel = aFile.getChannel();
            ByteBuffer buf = ByteBuffer.allocate(48);
            // 从FileChannel通道读取数据到缓冲区ByteBuffer
            int bytesRead = inChannel.read(buf);
            while (bytesRead != -1) {
                System.out.println("读取到的数据长度 " + bytesRead);
                buf.flip();
                while (buf.hasRemaining()) {
                    System.out.print((char) buf.get());
                }
                buf.clear();
                // 继续读取文件信息
                bytesRead = inChannel.read(buf);
            }
            aFile.close();
        } 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
    @Test
    public void fileChannelWrite() {

        try {
            // 开启FileChannel
            RandomAccessFile aFile = new RandomAccessFile("D:\job\git\java-study\nio\src\main\resources\fileChanelWrite.txt", "rw");
            FileChannel inChannel = aFile.getChannel();
            ByteBuffer buf = ByteBuffer.allocate(48);
            byte[] forWrite = "需要写入的字符串。".getBytes(StandardCharsets.UTF_8);
            buf.put(forWrite, 0, forWrite.length);
            buf.flip();
            // 写入数据
            while (buf.hasRemaining()) {
                inChannel.write(buf);
            }
            aFile.close();
        } 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
    /**
     * UDP 服务端
     */
    @Test
    public void datagramChannelService() {
        try {
            // 获取通道
            DatagramChannel datagramChannel = DatagramChannel.open();
            // 绑定端口8989,作为UDP服务端
            datagramChannel.bind(new InetSocketAddress(8989));
            // 分配Buffer,用于收发数据
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            while (true) {
                buffer.clear();
                // 等待接受客户端发送数据
                SocketAddress socketAddress = datagramChannel.receive(buffer);
                if (socketAddress != null) {
                    buffer.flip();
                    byte[] b = new byte[buffer.limit()];
                    int bufferReceiveIndex = 0;
                    while (buffer.hasRemaining()) {
                        b[bufferReceiveIndex++] = buffer.get();
                    }
                    System.out.println("收到客户端消息 " + socketAddress.toString() + ":" + new String(b, StandardCharsets.UTF_8));
                    // 接收到消息后给发送方回应
                    sendDataBack(socketAddress, datagramChannel);
                }
            }

        } catch (IOException e) {
            // ignore
        }
    }

    /**
     * 给socketAddress地址发送消息
     */
    private void sendDataBack(SocketAddress socketAddress, DatagramChannel datagramChannel) throws IOException {
        String message = "send back";
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        buffer.put(message.getBytes(StandardCharsets.UTF_8));
        buffer.flip();
        datagramChannel.send(buffer, socketAddress);
    }

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
    // UDP客户端
    @Test
    public void datagramChannelClient() {
        try {
            final DatagramChannel channel = DatagramChannel.open();
            // 开一个线程一直接收UDP服务端发送过来的消息
            new Thread(() -> {
                try {
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    while (true) {
                        buffer.clear();
                        SocketAddress socketAddress = channel.receive(buffer);
                        if (socketAddress != null) {
                            buffer.flip();
                            byte[] b = new byte[buffer.limit()];
                            int bufferReceiveIndex = 0;
                            while (buffer.hasRemaining()) {
                                b[bufferReceiveIndex++] = buffer.get();
                            }
                            System.out.println("收到消息 " + socketAddress.toString() + ":" + new String(b, StandardCharsets.UTF_8));
                        }
                    }
                } catch (Exception e) {
                    // ignore
                }

            }).start();

            int messageIndex = 0;
            // 控制台输入数据，然后发送给指定的地址
            while (true) {
                // 5S发送一次数据
                Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS);
                sendMessage(channel, new InetSocketAddress("192.168.5.14", 8989), String.valueOf(messageIndex++));
            }

        } catch (IOException e) {
            // ignore
        }
    }


    private void sendMessage(DatagramChannel channel, InetSocketAddress address, String mes) throws IOException {
        if (mes == null || mes.isEmpty()) {
            return;
        }
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        buffer.clear();
        buffer.put(mes.getBytes(StandardCharsets.UTF_8));
        buffer.flip();
        channel.send(buffer, address);
    }

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
    /**
     * UDP connect() 在特定的地址上收发消息
     */
    @Test
    public void datagramChannelConnect() {
        try {
            // 获取通道
            DatagramChannel datagramChannel = DatagramChannel.open();
            // 连接到特定的地址，time-a.nist.gov 获取时间。只在这个地址间收发消息 write,read 方法
            datagramChannel.connect(new InetSocketAddress("time-a.nist.gov", 37));
            ByteBuffer buffer = ByteBuffer.allocate(8);
            buffer.order(ByteOrder.BIG_ENDIAN);
            buffer.put((byte) 0);
            buffer.flip();
            // 发送数据到 time-a.nist.gov
            datagramChannel.write(buffer);
            buffer.clear();
            // 前四个字节补0
            buffer.putInt(0);
            // 从 time-a.nist.gov 读取数据
            datagramChannel.read(buffer);
            buffer.flip();
            // convert seconds since 1900 to a java.util.Date
            long secondsSince1900 = buffer.getLong();
            long differenceBetweenEpochs = 2208988800L;
            long secondsSince1970 = secondsSince1900 - differenceBetweenEpochs;
            long msSince1970 = secondsSince1970 * 1000;
            Date time = new Date(msSince1970);
            // 打印时间
            System.out.println(time);
        } catch (Exception e) {
            // ignore
        }
    }

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
    /**
     * TCP客户端,阻塞模式
     */
    @Test
    public void socketChannelClient() {
        try {
            SocketChannel channel = SocketChannel.open();
            // 这里使用的是阻塞模式
            channel.connect(new InetSocketAddress("192.168.5.14", 6800));
            // KEEP ALIVE setOption()函数的使用,一定要在连接成功之后设置
            channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            while (true) {
                buffer.clear();
                int readLength = channel.read(buffer);
                if (readLength >= 0) {
                    buffer.flip();
                    byte[] b = new byte[buffer.limit()];
                    int bufferReceiveIndex = 0;
                    while (buffer.hasRemaining()) {
                        b[bufferReceiveIndex++] = buffer.get();
                    }
                    System.out.println("收到消息 " + ":" + new String(b, StandardCharsets.UTF_8));
                    // 把收到的消息又发送回去
                    buffer.clear();
                    buffer.put(b);
                    buffer.flip();
                    channel.write(buffer);
                }

            }
        } 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
    /**
     * TCP服务端 -- 阻塞模式
     */
    @Test
    public void socketChannelServer() {
        try {
            ServerSocketChannel  channel = ServerSocketChannel.open();
            channel.bind(new InetSocketAddress("192.168.5.14", 6800));
            while (true) {
                // 接收客户端的连接,之后拿到的就是SocketChannel了，之后都是基于SocketChannel做相应的操作
                SocketChannel clientSocketChannel = channel.accept();
                clientSocketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                buffer.clear();
                buffer.put("hello".getBytes(StandardCharsets.UTF_8));
                buffer.flip();
                // 给客户端发送消息
                clientSocketChannel.write(buffer);
                // 在收下客户端的消息
                buffer.clear();
               int readLength = clientSocketChannel.read(buffer);
                if (readLength >= 0) {
                    buffer.flip();
                    byte[] b = new byte[buffer.limit()];
                    int bufferReceiveIndex = 0;
                    while (buffer.hasRemaining()) {
                        b[bufferReceiveIndex++] = buffer.get();
                    }
                    System.out.println("收到消息 " + ":" + new String(b, StandardCharsets.UTF_8));
                }
            }
        } 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
public abstract class Buffer {
    
    /**
     * 获取当前缓冲区的容量 -- capacity
     */
    public final int capacity();

    /**
     * 获取当前缓冲区的位置 -- position
     */
    public final int position();

    /**
     * 设置当前缓冲区的位置 -- position
     */
    public final Buffer position(int newPosition);

    /**
     * 获取当前缓冲区的限制 -- limit
     */
    public final int limit();

    /**
     * 设置当前缓冲区的限制 -- limit
     */
    public final Buffer limit(int newLimit);

    /**
     * mark(), reset()函数是配对使用的，将当前缓冲区的标记(mark)设置在当前位置(position) -- mark
     */
    public final Buffer mark();

    /**
     * 通过调用mark()方法，可以标记Buffer中的一个特定position。
     * 之后可以通过调用Buffer.reset()方法恢复到这个position
     */
    public final Buffer reset();

    /**
     * 	清除此缓存区。将position = 0;limit = capacity;mark = -1;一般在写入数据之前调用
     */
    public final Buffer clear();

    /**
     * flip()方法可以把Buffer从写模式切换到读模式。调用flip方法会把position归零，
     * 并设置limit为之前的position的值。
     * 也就是说，现在position代表的是读取位置，limit标示的是已写入的数据位置。
     */
    public final Buffer flip();

    /**
     * 将position设回0，这个时候你可以重读Buffer中的所有数据。
     * limit保持不变，仍然表示能从Buffer中读取多少个元素
     */
    public final Buffer rewind();

    /**
     * return limit - position; 返回limit和position之间相对位置差
     */
    public final int remaining();

    /**
     * return position < limit，返回是否还有未读内容
     */
    public final boolean hasRemaining();

    /**
     * 判断此缓冲区是否为只读
     */
    public abstract boolean isReadOnly();

    /**
     * 判断此缓冲区是否由可访问的数组支持
     */
    public abstract boolean hasArray();

    /**
     * 返回支持此缓冲区的数组
     */
    public abstract Object array();

    /**
     * 返回该缓冲区的缓冲区的第一个元素的背衬数组中的偏移量
     */
    public abstract int arrayOffset();

    /**
     * 判断个缓冲区是否为 direct
     */
    public abstract boolean isDirect();

}

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
    @Test
    public void byteBufferTest() {
        // 创建一个ByteBuffer实例
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        // 清空
        buffer.clear();
        // 写入数据
        byte[] putByteArray = "hello word!".getBytes(StandardCharsets.UTF_8);
        buffer.put(putByteArray);
        // 切换到读模式
        buffer.flip();
        // 把数据读取出来
        buffer.slice();
        while (buffer.hasRemaining()) {
            System.out.print((char) buffer.get());
        }
        System.out.println();
        // 重新读
        buffer.rewind();
        // ps: 这个时候buffer.limit()就是数组元素的个数
        byte[] retByte = new byte[buffer.limit()];
        buffer.get(retByte);
        System.out.println(new String(retByte, StandardCharsets.UTF_8));
    }

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
    @Test
    public void charBufferTest() {
        // 创建一个ByteBuffer实例
        CharBuffer buffer = CharBuffer.allocate(1024);
        // 清空
        buffer.clear();
        // 写入数据
        char[] putArray = "hello word!".toCharArray();
        buffer.put(putArray);
        // 切换到读模式
        buffer.flip();
        // 把数据读取出来
        buffer.slice();
        while (buffer.hasRemaining()) {
            System.out.print(buffer.get());
        }
        System.out.println();
        // 重新读
        buffer.rewind();
        // ps: 这个时候buffer.limit()就是数组元素的个数
        char[] retByte = new char[buffer.limit()];
        buffer.get(retByte);
        System.out.println(String.valueOf(retByte));
    }

1
2
Selector selector = Selector.open();

1
2
3
channel.configureBlocking(false);
SelectionKey key = channel.register(selector, Selectionkey.OP_READ);

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
    @Test
    public void tcpClient() {
        try {
            SocketChannel socketChannel = SocketChannel.open();
            // 连接
            socketChannel.connect(new InetSocketAddress("192.168.5.14", 6800));
            ByteBuffer writeBuffer = ByteBuffer.allocate(32);
            ByteBuffer readBuffer = ByteBuffer.allocate(32);
            writeBuffer.put("hello".getBytes());
            writeBuffer.flip();
            while (true) {
                writeBuffer.rewind();
                socketChannel.write(writeBuffer);
                readBuffer.clear();
                socketChannel.read(readBuffer);
                readBuffer.flip();
                byte[] b = new byte[readBuffer.limit()];
                int bufferReceiveIndex = 0;
                while (readBuffer.hasRemaining()) {
                    b[bufferReceiveIndex++] = readBuffer.get();
                }
                System.out.println("received : " + new String(b));
            }
        } catch (Exception e) {
            // ignore
        }
    }


    @Test
    public void tcpServer() {

        try {
            // 创建一个ServerSocketChannel通道
            ServerSocketChannel serverChannel = ServerSocketChannel.open();
            // 绑定6800端口
            serverChannel.bind(new InetSocketAddress("192.168.5.14", 6800));
            // 设置非阻塞
            serverChannel.configureBlocking(false);
            // Selector创建
            Selector selector = Selector.open();
            // 注册 channel，并且指定感兴趣的事件是 Accept
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);
            ByteBuffer readBuff = ByteBuffer.allocate(1024);
            ByteBuffer writeBuff = ByteBuffer.allocate(1024);
            writeBuff.put("received".getBytes());
            writeBuff.flip();
            while (true) {
                if (selector.select() > 0) {
                    Set<SelectionKey> readyKeys = selector.selectedKeys();
                    Iterator<SelectionKey> readyKeyIterator = readyKeys.iterator();
                    while (readyKeyIterator.hasNext()) {
                        SelectionKey key = readyKeyIterator.next();
                        readyKeyIterator.remove();

                        if (key.isAcceptable()) {
                            // 连接
                            ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                            SocketChannel socketChannel = serverSocketChannel.accept();
                            socketChannel.configureBlocking(false);
                            // 我们又给注册到Selector里面去了，声明这个channel只对读操作感兴趣。
                            socketChannel.register(selector, SelectionKey.OP_READ);
                        } else if (key.isReadable()) {
                            // 读
                            SocketChannel socketChannel = (SocketChannel) key.channel();
                            readBuff.clear();
                            socketChannel.read(readBuff);
                            readBuff.flip();
                            byte[] b = new byte[readBuff.limit()];
                            int bufferReceiveIndex = 0;
                            while (readBuff.hasRemaining()) {
                                b[bufferReceiveIndex++] = readBuff.get();
                            }
                            System.out.println("received : " + new String(b));
                            // 修改selector对channel感兴趣的事件
                            key.interestOps(SelectionKey.OP_WRITE);
                        } else if (key.isWritable()) {
                            // 写
                            writeBuff.rewind();
                            SocketChannel socketChannel = (SocketChannel) key.channel();
                            socketChannel.write(writeBuff);
                            // 修改selector对channel感兴趣的事件
                            key.interestOps(SelectionKey.OP_READ);
                        }
                    }
                }
            }

        } catch (IOException e) {
            // ignore
        }

    }