Java NIO系列 - Channel

Posted 2021-01-08 零壹技术栈

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Java NIO系列 - Channel相关的知识，希望对你有一定的参考价值。

前言

上文讲到Java NIO一些基本概念。在标准的IO中，都是基于字节流/字符流进行数据操作的，而在NIO中则是是基于Channel和Buffer进行操作，其中的Channel的虽然模拟了流的概念，实则大不相同。

本文将详细阐述NIO中的通道Channel的概念和具体的用法。

Channel和Stream的区别

区别	Stream	Channel
是否支持异步	不支持	支持
是否支持双向数据传输	不支持，只能单向	支持，既可以从通道读取数据，也可以向通道写入数据
是否结合Buffer使用	不	必须结合Buffer使用
性能	较低	较高

Channel用于在字节缓冲区和位于通道另一侧的服务（通常是文件或者套接字）之间以便有效的进行数据传输。借助通道，可以用最小的总开销来访问操作系统本身的I/O服务。

需要注意的是Channel必须结合Buffer使用，应用程序不能直接向通道中读/写数据，也就是缓冲区充当着应用程序和通道数据流动的转换的角色。

正文

Channel的源码

查看Channel的源码。所有的接口都实现于Channel接口，从接口上来看，所有的通道都有这两种操作：检查通道的开启状态和关闭通道。

public interface Channel extends Closeable {
    public boolean isOpen();

    public void close() throws IOException;
}

Channel的分类

广义上来说通道可以被分为两类：文件I/O和网络I/O，也就是文件通道和套接字通道。如果分的更细致一点则是：

FileChannel：从文件读写数据；
SocketChannel：通过TCP读写网络数据；
ServerSocketChannel：可以监听新进来的TCP连接，并对每个链接创建对应的SocketChannel；
DatagramChannel：通过UDP读写网络中的数据。

Channel的特性

单向or双向

通道既可以是单向的也可以是双向的。只实现ReadableByteChannel接口中的read()方法或者只实现WriteableByteChannel接口中的write()方法的通道皆为单向通道，同时实现ReadableByteChannel和WriteableByteChannel为双向通道，比如ByteChannel。

1 2	public interface ByteChannel extends ReadableByteChannel, WritableByteChannel { }

对于Socket通道来说，它们一直是双向的，而对于FileChannel来说，它同样实现了ByteChannel，但是通过FileInputStream的getChannel()获取的FileChannel只具有文件的只读权限。

注意：调用FileChannel的write()方法会抛出了NonWriteChannelException异常。

阻塞or非阻塞

通道的工作模式有两种：阻塞或非阻塞。在非阻塞模式下，调用的线程不会休眠，请求的操作会立刻返回结果；在阻塞模式下，调用的线程会产生休眠。

除FileChannel不能运行在非阻塞模式下，其余的通道都可阻塞运行也可以以非阻塞的方式运行。

另外从SelectableChannel引申出的类可以和支持有条件选择的Selector结合使用，进而充分利用多路复用的I/O(Multiplexed I/O)来提高性能。

SelectableChannel的源码中有以下几个抽象方法，可以看出支持配置两种工作模式：

public abstract class SelectableChannel extends AbstractInterruptibleChannel implements Channel {
    /**
     * 配置是否为Channel阻塞模式
     */
    public abstract SelectableChannel configureBlocking(boolean block) throws IOException;
    /**
     * 判断是否为Channel阻塞模式
     */
    public abstract boolean isBlocking();
    /**
     * 获取阻塞的锁对象
     */
    public abstract Object blockingLock();
}

对于Socket通道类来说，通常与Selector共同使用以提高性能。需要注意的是通道不能被同时使用，一个打开的通道代表着与一个特定I/O服务进行连接并封装了该连接的状态，通道一旦关闭，该连接便会断开。

通道的close()比较特殊，无论在通道时在阻塞模式下还是非阻塞模式下，由于close()方法的调用而导致底层I/O的关闭都可能会造成线程的暂时阻塞。在一个已关闭的通道上调用close()并没有任何意义，只会立即返回。

Channel的实战

对于Socket通道来说存在直接创建新Socket通道的方法，而对于文件通道来说，升级之后的FileInputStream、FileOutputStream和RandomAccessFile提供了getChannel()方法来获取通道。

FileChannel

Java NIO中的FileChannel是一个连接到文件的通道，可以通过文件通道读写文件。文件通道总是阻塞式的，因此FileChannel无法设置为非阻塞模式。

文件读写

(一). 文件写操作：

public static void testWriteOnFileChannel() {
    try {
        RandomAccessFile randomAccess = new RandomAccessFile("D://test.txt", "rw");
        FileChannel fileChannel = randomAccess.getChannel();

        byte[] bytes = new String("Java Non-blocking IO").getBytes();
        ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);

        // 将缓冲区中的字节写入文件通道中
        fileChannel.write(byteBuffer);
        // 强制将通道中未写入磁盘的数据立刻写入到磁盘
        fileChannel.force(true);
        // 清空缓冲区，释放内存
        byteBuffer.clear();
        fileChannel.close();
    } catch (FileNotFoundException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

(二). 文件读操作：

public static void testReadOnFileChannel() {
    try {
        FileInputStream inputStream = new FileInputStream(new File("D://test.txt"));
        FileChannel fileChannel = inputStream.getChannel();

        ByteBuffer byteBuffer = ByteBuffer.allocate(10);
        // 不断地写入缓冲区，写一次读一次
        while (fileChannel.read(byteBuffer) != -1) {
            // 缓冲区从写模式切换为读模式
            byteBuffer.flip();
            // 开始读取
            while (byteBuffer.hasRemaining()) {
                // 一个字节一个字节地读取，并向后移动position地位置
                System.out.print((char) byteBuffer.get());
            }
            // 缓冲区不会被自动覆盖，需要主动调用该方法(实际上还是覆盖)
            byteBuffer.clear();
        }
        fileChannel.close();
    } catch (FileNotFoundException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

文件读写测试：

public static void main(String[] args) {
    System.out.println("Start to write");
    // 通过FileChannel写入数据
    testWriteOnFileChannel();

    System.out.println("Start to read");
    // 通过FileChannel读取数据
    testReadOnFileChannel();
}

测试结果：

transferFrom和transferTo

(一). transferFrom()的使用

FileChannel的transferFrom()方法可以将数据从源通道传输到FileChannel中。下面是一个简单的例子：

public static void testTransferFrom(){
    try {
        RandomAccessFile fromFile = new RandomAccessFile("D://file1.txt", "rw");
        FileChannel fromChannel = fromFile.getChannel();
        RandomAccessFile toFile = new RandomAccessFile("D://file2.txt", "rw");
        FileChannel toChannel = toFile.getChannel();

        long position = 0;
        long count = fromChannel.size();
        toChannel.transferFrom(fromChannel, position, count);
    } catch (FileNotFoundException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

(二). transferTo()的使用

transferTo()方法将数据从FileChannel传输到目标channel中。下面是一个简单的例子：

public static void testTransferTo() {
    try {
        RandomAccessFile fromFile = new RandomAccessFile("D://file1.txt", "rw");
        FileChannel fromChannel = fromFile.getChannel();
        RandomAccessFile toFile = new RandomAccessFile("D://file3.txt", "rw");
        FileChannel toChannel = toFile.getChannel();

        long position = 0;
        long count = fromChannel.size();
        fromChannel.transferTo(position, count, toChannel);
    } catch (FileNotFoundException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    }

ServerSocketChannel

Java NIO中的ServerSocketChannel是一个可以监听新进来的TCP连接的通道。它类似ServerSocket，要注意的是和DatagramChannel和SocketChannel不同，ServerSocketChannel本身不具备传输数据的能力，而只是负责监听传入的连接和创建新的SocketChannel。

ServerSocketChannel的用法

(一). 创建ServerSocketChannel

通过ServerSocketChannel.open()方法来创建一个新的ServerSocketChannel对象，该对象关联了一个未绑定ServerSocket的通道。通过调用该对象上的socket()方法可以获取与之关联的ServerSocket。

1	ServerSocketChannel socketChannel = ServerSocketChannel.open();

(二). 为ServerSocketChannel绑定监听端口号

在JDK 1.7之前，ServerSocketChannel没有bind()方法，因此需要通过他关联的的socket对象的socket()来绑定。

1 2	// JDK1.7之前 serverSocketChannel.socket().bind(new InetSocketAddress(25000));

从JDK1.7及以后，可以直接通过ServerSocketChannel的bind()方法来绑定端口号。

1 2	// JDK1.7之后 serverSocketChannel.bind(new InetSocketAddress(25000));

(三). 设置ServerSocketChannel的工作模式

ServerSocketChannel底层默认采用阻塞的工作模式，它提供了一个configureBlocking()方法，允许配置ServerSocketChannel以非阻塞方式运行。

1 2	// 设置为非阻塞模式 serverSocketChannel.configureBlocking(false);

进一步查看configureBlocking源码如下：

public final SelectableChannel configureBlocking(boolean block) throws IOException {
    synchronized (regLock) {
        if (!isOpen())
            throw new ClosedChannelException();
        if (blocking == block)
            return this;
        if (block && haveValidKeys())
            throw new IllegalBlockingModeException();
        implConfigureBlocking(block);
        blocking = block;
    }
    return this;
}

Javadoc解释configureBlocking()方法用于调整底层通道的工作模式，即阻塞和非阻塞，默认是阻塞工作模式。

如果block设置为true，直接返回当前的阻塞式的通道；如果block设置为false，configureBlocking()方法会调用implConfigureBlocking()方法。这里implConfigureBlocking()是由ServerSocketChannelImpl实现，最终调用了IOUtil中的native方法configureBlocking()。

(四). 监听新进来的连接

通过ServerSocketChannel.accept()方法监听新进来的连接，这里需要根据configureBlocking()的配置区分两种工作模式的使用：

在阻塞模式下，当accept()方法返回的时候，它返回一个包含新连接的SocketChannel，否则accept()方法会一直阻塞到有新连接到达。
在非阻塞模式下，在没有新连接的情况下，accept()会立即返回null，该模式下通常不会仅仅监听一个连接，因此需在while循环中调用accept()方法.

阻塞模式：

while(true) {
    SocketChannel socketChannel = serverSocketChannel.accept();
    // 新连接没到达之前，后面的程序无法继续执行
    InetSocketAddress remoteAddress = (InetSocketAddress) socketChannel.getRemoteAddress();
    // 其他操作
}

非阻塞模式：

while(true) {
    SocketChannel socketChannel = serverSocketChannel.accept();
    // 新连接没到达之前，后面程序一直循环，直到检测到socketChannel不为null时进入真正的执行逻辑
    if(socketChannel != null) {
        InetSocketAddress remoteAddress = (InetSocketAddress) socketChannel.getRemoteAddress();
        // 其他操作
    }
}

(五). 关闭ServerSocketChannel

通过调用ServerSocketChannel.close()方法来关闭ServerSocketChannel。

1	serverSocketChannel.close();

ServerSocketChannel的完整示例

(一). 阻塞模式

代码示例：

public static void blockingTest() throws IOException {
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    serverSocketChannel.bind(new InetSocketAddress(25000));

    System.out.println("ServerSocketChannel listening on 25000...");

    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

    while(true) {
        SocketChannel socketChannel = serverSocketChannel.accept();
        InetSocketAddress remoteAddress = (InetSocketAddress) socketChannel.getRemoteAddress();
        System.out.println("Remote address: " + remoteAddress.getHostString());

        while (socketChannel.read(byteBuffer) != -1) {
            byteBuffer.flip();
            while (byteBuffer.hasRemaining()) {
                System.out.print((char) byteBuffer.get());
            }
            byteBuffer.clear();
        }
    }
}

运行结果：

(二). 非阻塞模式

代码示例：

public static void nonBlockingTest() throws IOException {
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    serverSocketChannel.configureBlocking(false);
    serverSocketChannel.bind(new InetSocketAddress(25001));
    System.out.println("ServerSocketChannel listening on 25001...");

    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    while (true) {
        SocketChannel socketChannel = serverSocketChannel.accept();
        System.out.println("SocketChannel： " + socketChannel);
        if (socketChannel != null) {
            InetSocketAddress remoteAddress = (InetSocketAddress) socketChannel.getRemoteAddress();
            System.out.println("Remote address: " + remoteAddress.getHostString());

            while (socketChannel.read(byteBuffer) != -1) {
                byteBuffer.flip();
                while (byteBuffer.hasRemaining()) {
                    System.out.print((char) byteBuffer.get());
                }
                byteBuffer.clear();
            }
        }
    }
}

运行结果：

SocketChannel

Java NIO中的SocketChannel是一个连接到TCP网络套接字的通道，它是Socket类的对等类。

通常SocketChannel在客户端向服务器发起连接请求，每个SocketChannel对象创建时都关联一个对等的Socket对象。同样SocketChannel也可以运行在非阻塞模式下。

SocketChannel的用法

SocketChannel创建的方式有两种：

客户端主动创建：客户端打开一个SocketChannel并连接到某台服务器上；
服务端被动创建：一个新连接到达ServerSocketChannel时，服务端会创建一个SocketChannel。

(一). 创建SocketChannel

通过SocketChannel的静态方法open()创建SocketChannel对象。此时通道虽然打开，但并未建立连接。此时如果进行I/O操作会抛出NotYetConnectedException异常。

1	SocketChannel socketChannel = SocketChannel.open();

(二). 连接指定服务器

通过SocketChannel对象的connect()连接指定地址。该通道一旦连接，将保持连接状态直到被关闭。可通过isConnected()来确定某个SocketChannel当前是否已连接。

阻塞模式：

如果在客户端的SocketChannel阻塞模式下，即服务器端的ServerSocketChannel也为阻塞模式：

1
2
3

socketChannel.connect(new InetSocketAddress("127.0.0.1", 25000));
// connect()方法调用以后，socketChannel底层的连接创建完成后，才会执行后面的打印语句
System.out.println("连接创建完成...");

非阻塞模式：

两点需要注意：其一，SocketChannel需要通过configureBlocking()设置为非阻塞模式；其二，非阻塞模式下，connect()方法调用后会异步返回，为了确定连接是否建立，需要调用finishConnect()的方法。

socketChannel.configureBlocking(false);
socketChannel.connect(new InetSocketAddress("127.0.0.1", 25001));
// connect()方法调用以后，异步返回，需要手动调用finishConnect确保连接创建

while(!socketChannel.finishConnect()){
    // 检测到还未创建成功则睡眠10ms
    TimeUnit.MILLISECONDS.sleep(10);
}
System.out.println("连接创建完成...");

(三). 从SocketChannel读数据

利用SocketChannel对象的read()方法将数据从SocketChannel读取到Buffer。

ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

// 非阻塞模式下，read()方法在尚未读取到任何数据时可能就返回了，所以需要关注它的int返回值。
while (socketChannel.read(byteBuffer) != -1) {
    byteBuffer.flip();
    while (byteBuffer.hasRemaining()) {
        System.out.println((char) byteBuffer.get());
    }
    byteBuffer.clear();
}

(四). 向SocketChannel写数据

利用SocketChannel对象的write()将Buffer的数据写入SocketChannel。

ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
byteBuffer.put("Client Blocking SocketChannel".getBytes());
// byteBuffer.put("Client Non-Blocking SocketChannel".getBytes());
byteBuffer.flip();

// 非阻塞模式下，write()方法在尚未写出任何内容时可能就返回了。所以需要在循环中调用write()
while (byteBuffer.hasRemaining()) {
    socketChannel.write(byteBuffer);
}

// 保持睡眠，观察控制台输出
TimeUnit.SECONDS.sleep(20000);
socketChannel.close();

(五). 关闭SocketChannel

利用SocketChannel对象的close()方法关闭SocketChannel。

1	socketChannel.close();

SocketChannel的完整示例

(一). 阻塞模式

代码示例：

public static void blockingWrite() throws Exception {
    SocketChannel socketChannel = SocketChannel.open();
    socketChannel.connect(new InetSocketAddress("127.0.0.1", 25000));

    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    byteBuffer.put("Client Blocking SocketChannel".getBytes());
    byteBuffer.flip();

    while (byteBuffer.hasRemaining()) {
        socketChannel.write(byteBuffer);
    }

    TimeUnit.SECONDS.sleep(20000);
    socketChannel.close();
}

服务端打印结果：

(一). 非阻塞模式

代码示例：

public static void nonBlockingWrite() throws Exception {
    SocketChannel socketChannel = SocketChannel.open();
    socketChannel.configureBlocking(false);
    socketChannel.connect(new InetSocketAddress("127.0.0.1", 25001));

    while(!socketChannel.finishConnect()){
        TimeUnit.MILLISECONDS.sleep(10);
    }

    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    byteBuffer.put("Client Non-Blocking SocketChannel".getBytes());
    byteBuffer.flip();
    while (byteBuffer.hasRemaining()) {
        socketChannel.write(byteBuffer);
    }

    TimeUnit.SECONDS.sleep(20000);
    socketChannel.close();
}

服务端打印结果：

DatagramChannel

Java NIO中的DatagramChannel是一个能收发UDP包的通道，其底层实现为DatagramSocket + Selector。DatagramChannel可以调用socket()方法获取对等DatagramSocket对象。
DatagramChannel对象既可以充当服务端（监听者），也可以充当客户端（发送者）。如果需要新创建的通道负责监听，那么该通道必须绑定一个端口（或端口组）：

DatagramChannel的完整示例

数据报发送方：

public static void main(String[] args) throws Exception {
    DatagramChannel datagramChannel = DatagramChannel.open();
    ByteBuffer byteBuffer = ByteBuffer.wrap("DatagramChannel Sender".getBytes());
    int byteSent = datagramChannel.send(byteBuffer, new InetSocketAddress("127.0.0.1", 50020));
    System.out.println("Byte sent is: " + byteSent);
}

数据报接收方：

public static void main(String[] args) throws Exception {
    DatagramChannel datagramChannel = DatagramChannel.open();
    datagramChannel.socket().bind(new InetSocketAddress(50020));

    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    datagramChannel.receive(byteBuffer);
    byteBuffer.flip();

    while (byteBuffer.hasRemaining()) {
        System.out.print((char) byteBuffer.get());
    }
}

先运行DatagramChannelReceiveTest，再运行DatagramChannelSendTest，观察控制台输出：

数据报发送方：

数据报接收方：

工具类Channels

NIO通道提供了一个便捷的通道类Channels，其中定义了几种静态的工厂方法以简化通道和流转换。其中常用的方法如下：

方法	返回	描述
newChannel(InputStream in)	ReadableByteChannel	返回一个将从给定的输入流读取数据的通道。
newChannel(OutputStream out)	WritableByteChannel	返回一个将向给定的输出流写入数据的通道。
newInputStream(ReadableByteChannel ch)	InputStream	返回一个将从给定的通道读取字节的流。
newOutputStream(WritableByteChannel ch)	OutputStream	返回一个将向给定的通道写入字节的流。
newReader(ReadableByteChannel ch, CharsetDecoder dec, int minBufferCap)	Reader	返回一个reader，它将从给定的通道读取字节并依据提供的字符集名称对读取到的字节进行解码。
newReader(ReadableByteChannel ch, String csName)	Reader	返回一个reader，它将从给定的通道读取字节并依据提供的字符集名称将读取到的字节解码成字符。
newWriter(WritableByteChannel ch, CharsetEncoder dec, int minBufferCap)	Writer	返回一个writer，它将使用提供的字符集名称对字符编码并写到给定的通道中。
newWriter(WritableByteChannel ch, String csName)	Writer	返回一个writer，它将依据提供的字符集名称对字符编码并写到给定的通道中。

总结

本文针对NIO中的通道的做了详细的介绍，对于文件通道FileChannel，网络通道SocketChannel、ServerSocketChannel和DatagramChannel进行了实战演示。

篇幅较长，可见NIO提供的原生的通道API在使用上并不是太容易。

欢迎扫码关注我的个人技术公众号：零壹技术栈

本帐号将持续分享后端技术干货，包括虚拟机基础，多线程编程，高性能框架，异步、缓存和消息中间件，分布式和微服务，架构学习和进阶等学习资料和文章。

以上是关于Java NIO系列 - Channel的主要内容，如果未能解决你的问题，请参考以下文章

Java NIO系列教程 Channel

Java NIO系列教程 Scatter/Gather

（四：NIO系列） Java NIO Selector

二.Netty入门到超神系列-Java NIO 三大核心(selector,channel,buffer)

Java NIO系列教程 Scatter/Gather

Java NIO系列 - Selector