JAVA NIO 之Channel

Posted 2021-01-13 tangquanbin

缓冲区本质上是一块可以写入数据，然后可以从中读取数据的内存。Channel 通道就是将数据传输给 ByteBuffer 对象或者从 ByteBuffer 对象获取数据进行传输。

    Channel 用于在字节缓冲区和位于通道另一侧的实体（通常是一个文件或套接字）之间有效地传输数据。常用Channel有FileChannel、SocketChannel、DatagramChannel、ServerSocketChannel
Socket 可以通过socket 通道的工厂方法直接创建。但是FileChannel 对象却只能通过在一个打开的 RandomAccessFile、FileInputStream 或 FileOutputStream对象上调用 getChannel( )方法来获取。
通道可以以阻塞（blocking）或非阻塞（nonblocking）模式运行。只有面向流的（stream-oriented）的通道，如 sockets 和 pipes 才能使用非阻塞模式。

1.打开通道

a.打开FileChannel，以RandomAccessFile为例 ：

            RandomAccessFile aFile = new RandomAccessFile("G:\\we.txt", "rw");
            FileChannel inChannel = aFile.getChannel();

 b.打开SocketChannel：

SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("127.0.0.1",9011));

 c.打开ServerSocketChannel

ServerSocketChannel ssc = ServerSocketChannel.open( );
ssc.socket( ).bind (new InetSocketAddress (9011));

 2.使用通道

a.使用FileChannel从指定文件中读取数据：

/**
 * @author tangquanbin
 * @date 2018/10/18 21:00
 */
public class ChannelTest {

    public static void main(String[] args) {

        //RandomAccessFile、FileInputStream 或 FileOutputStream的close()会同时关闭chanel。
        
        try (RandomAccessFile aFile = new RandomAccessFile("F:\\test.txt", "rw")) {
            //1.FileChannel 对象却只能通过在一个打开的 RandomAccessFile、FileInputStream 或 FileOutputStream对象上调用 getChannel( )方法来获取
            //2.三个 socket 通道类：SocketChannel、ServerSocketChannel 和 DatagramChannel有可以直接创建新 socket 通道的工厂方法
            //随机流, mode 的值可选 "r"：可读，"w" ：可写，"rw"：可读写；
            FileChannel inChannel = aFile.getChannel();

            ByteBuffer buf = ByteBuffer.allocate(48);

            int bytesRead = inChannel.read(buf);
            while (bytesRead != -1) {
                //通过flip()方法将Buffer从写模式切换到读模式
                buf.flip();
                while (buf.hasRemaining()) {
                    System.out.print((char) buf.get());
                }
                //两种方式能清空缓冲区：调用clear()或compact()方法,clear()方法会清空整个缓冲区。compact()方法只会清除已经读过的数据
                buf.clear();
                bytesRead = inChannel.read(buf);
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

　b.在两个通道中复制数据：

/**
 * 在通道之间复制数据
 * @author tangquanbin
 * @date 2018/10/18 21:12
 */
public class ChannelCopy {

    public static void main(String[] argv) throws IOException {
        FileInputStream inputStream = new FileInputStream("G:\\test.txt");
        FileChannel inChanel = inputStream.getChannel();
        FileOutputStream outputStream = new FileOutputStream("G:\\copyTest.txt");
        FileChannel outChanel = outputStream.getChannel();
        //channelCopy1 channelCopy2方法都是两个通道复制数据
        channelCopy1(inChanel, outChanel);
//        channelCopy2 (inChanel, outChanel);
        inChanel.close();
        outChanel.close();
    }

    /**
     *
     * @param inChanel
     * @param outChanel
     * @throws IOException
     */
    private static void channelCopy1(FileChannel inChanel, FileChannel outChanel) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocateDirect(16 * 1024);

        //将数据从输入通道读入缓冲区
        while (inChanel.read(buffer) != -1) {
            // buffer切换到读模式
            buffer.flip();

            while (buffer.hasRemaining()) {
                //如果buffer中有数据就将数据写入输出通道
                outChanel.write(buffer);
            }
            // buffer切换到写模式，确保缓冲区为空，准备继续填充数据
            buffer.clear();
        }
    }

    private static void channelCopy2(FileChannel inChanel, FileChannel outChanel) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocateDirect(16 * 1024);
        while (inChanel.read(buffer) != -1) {
            // Prepare the buffer to be drained
            buffer.flip();
            // Write to the channel; may block
            outChanel.write(buffer);
            // If partial transfer, shift remainder down
            // If buffer is empty, same as doing clear( )
            buffer.compact();
        }
        // EOF will leave buffer in fill state
        buffer.flip();
        // Make sure that the buffer is fully drained
        while (buffer.hasRemaining()) {
            outChanel.write(buffer);
        }
    }

　c.注意：一个连接到只读文件的 Channel 实例不能进行写操作，即使该实例所属的类可能有 write( )方法；FileChannel 实现 ByteChannel 

下面是ByteChannel的源码：

public interface ByteChannel
    extends ReadableByteChannel, WritableByteChannel
{

}

 可以看到ByteChannel其实什么也不做，这样设计起到了聚集的作用。

 FileChannel 接口既有write也有read方法，为何说不能写呢？那是当使用只读流获取通道时：

/**
 * @author tangquanbin
 * @date 2018/10/18 21:15
 */
public class ChannelOnlyReadTest {
    public static void main(String[] args) {
        try {
            ByteBuffer buffer = ByteBuffer.allocate(48);
            // 但是从 FileInputStream 对象的getChannel( )方法获取的 FileChannel 对象是只读的，因为 FileInputStream 对象总是以 read-only 的权限打开文件
            FileInputStream input = new FileInputStream ("G:\\we.txt");
            FileChannel channel = input.getChannel( );
            // 抛出 java.nio.channels.NonWritableChannelException
            channel.write (buffer);
            //FileInputStream的close同时会关闭chanel
            input.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

　上面的代码运行会报错：java.nio.channels.NonWritableChannelException

    因为：从 FileInputStream 对象的getChannel( )方法获取的 FileChannel 对象是只读的，因为 FileInputStream 对象总是以 read-only 的权限打开文件。所以没有写权限。

  之前我们用ServerSocket创建服务器的栗子中服务器是阻塞的。那NIO怎么实现非阻塞呢？

 上面讲的FileChannel 是非阻塞的，因为没有依靠SelectableChannel 超类。全部 socket 通道类（DatagramChannel、 SocketChannel 和 ServerSocketChannel） 都可以设置成非阻塞的。
  这里就要用到SelectableChannel 的方法

public abstract SelectableChannel configureBlocking(boolean block)
        throws IOException;

　　通道设置configureBlocking方法为false就表示非阻塞模式。

 下面是一个客服端和服务器代码说明；

 客服端：

/**
 * @author tangquanbin
 * @date 2018/10/18 21:08
 */
public class ScoketChannelClient {

    public static void main(String[] args) {

        SocketChannel socketChannel = null;

        try {
            ByteBuffer buffer = ByteBuffer.wrap ("客服端".getBytes());
            //获取SocketChannel
            socketChannel = SocketChannel.open();
            //设置通道为非阻塞
            socketChannel.configureBlocking(false);
            //连接
            socketChannel.connect(new InetSocketAddress("127.0.0.1",9011));

            //connect(InetSocketAddress) 等价于 open(InetSocketAddress)
            //SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1",9010));

            //一些判断
            if (!socketChannel.isBlocking()){
                System.out.println("非阻塞");
            }
            while (!socketChannel.finishConnect()){
                System.out.println("连接没有建立");
            }
            System.out.println("连接已建立");
            if (socketChannel.isConnected()){
                System.out.println("连接已建立");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                assert socketChannel != null;
                socketChannel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

　服务器：

package com.nio.java;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

/**
 * ServerSocketChannel
 * @author tangquanbin
 * @date 2018/10/18 21:29
 */
public class ChannelAccept
{
    public static final String HELLO = "Hello I am server.
";
    public static void main (String [] argv)throws Exception{
        int port = 9011;
        if (argv.length > 0) {
            port = Integer.parseInt (argv [0]);
        }
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open( );
        serverSocketChannel.socket( ).bind (new InetSocketAddress (port));
        //非阻塞模式
        serverSocketChannel.configureBlocking (false);
        while (true) {
            System.out.println ("Waiting for connections");
            //如果以非阻塞模式被调用，当没有传入连接在等待时，ServerSocketChannel.accept( )会立即返回 null
            SocketChannel sc = serverSocketChannel.accept( );
            if (sc == null) {
                // 没有连接。。。睡觉
                Thread.sleep (2000);
            } else {
                System.out.println ("Incoming connection from: "+ sc.socket().getRemoteSocketAddress( ));
                //wrap buffer的另一种创建方式
                ByteBuffer buffer = ByteBuffer.wrap (HELLO.getBytes( ));
                //可以使用 rewind()后退，重读已经被flip()的缓冲区中的数据。
                buffer.rewind( );
                sc.write (buffer);
            }
        }
    }
}

　Scatter/Gather（ ScatteringByteChannel 和 GatheringByteChannel）
    矢量化的 I/O 使您可以在多个缓冲区上自动执行一个 I/O 操作。使用多个而不是单个缓冲区来保存数据的读写方法。处理复杂数据结构时用。