Java NIO-非阻塞I/O（二）

Bertha 。 2024-03-16 11:57 43阅读 0赞

#### 文章目录 ####

*  1. SocketChannel
 *  2. ServerSocketChannel
 *  2. Channels类
 *  3. 异步通道（Java 7）
 *  4. Socket选项
 *  5. 就绪选择
 *  6. Selector类
 *  7. SelectionKey类

## 1. SocketChannel ##

通道将缓冲区的数据块移入或移出到个汇总给你I/O，如文件、Socket、数据报等。通道类的层次结构相当复杂，有多个接口和许多可选的操作，不过对于网络编程来说，实际上只有3个重要的通道类 ：`SocketChannel、ServerSocketChannel和DatagramChannel`。对于到目前为止谈到的TCP连接，只需要前连个通道类。

> Java的通道（Channel）与原始的流（Stream）相比，有几个显著的区别：
> 
>  *  双向性：原始的输入流和输出流都是单向的，一个流只能用于读取或写入。但是，通道是双向的。通过单个通道，既可以读取数据又可以写入数据。某些通道类型，如 FileChannel，可以同时进行读写，而网络通道可能由于协议的限制，例如TCP，其读写过程会发生在不同的套接字中，虽然逻辑上仍被认为是单一通道。
>  *  非阻塞I/O：流I/O是阻塞的，例如，当一个线程调用 read() 时，它会被阻塞，直到有一些数据可读。当一个线程调用 write() 时，它会被阻塞，直到数据被完全写入。然而，非阻塞I/O操作允许线程同时进行读或写请求的发起和其他操作，例如，当你向一个 SocketChannel 写数据时，如果网络缓冲区已满，write() 方法会在写完所有数据前返回。这使得线程能够处理其他任务。
>  *  通道之间的数据传输：如果两个通道都是文件通道，那么可以直接将数据从一个通道传输到另一个通道，而无需通过一个中间缓冲区来传输数据。这种方法可以非常高效地进行数据传输。
>  *  使用缓冲区进行数据传输：在新的I/O库中，所有数据都是通过缓冲区处理的。应用程序总是先向缓冲区写入数据，然后再将缓冲区写入通道，或者从通道读取数据到缓冲区，再从缓冲区读取数据。使用缓冲区的好处是可以提供更多的控制，并且可以提高效率。
>  *  选择器（Selectors）：Java NIO引入了选择器的概念，这是一个对象，可以通过它来监控多个通道的状态（例如，是否可以读，是否可以写等）。因此，单个线程可以管理多个通道，也就是说，管理多个网络连接。

SocketChannel类可以读写TCP socket。数据必须编码到ByteBuffer对象中完成读/写。每个SocketChannel都与一个对等端Socket对象关联，这个Socket可以用于高级配置，但有些应用采用默认选项就可以运行，对于这些应用程序，可以忽略这个需求

*  连接

SocketChannel类没有任何公共的构造函数，时间上，要使用两个静态的Open方法来创建新的SocketChannel对象

public static SocketChannel open(SocketAddress remote) throws IOException
    public static SocketChannel open() throws IOException

第一个方法会建立连接，这个方法会在连接成功或抛出异常之前阻塞。第二个方法不会建立连接，必须用connect方法进行连接（若我们在连接之前需要配置的时候，通常使用第二种方法）。使用非阻塞通道时，connect()方法会立即返回，甚至在建立连接之前就会返回，然后在等待操作系统建立连接时，我们可以做其他事。不过，程序在实际使用连接之前，必须调用 finishConnect()，这只是非阻塞模式需要的。对于阻塞模式 finishConnect()该方法会立即返回true。如果连接没有建立该方法返回false否则返回true。如果程序想检查是否连接完成，用下面方法：

//连接打开返回true
    public abstract boolean isConnected()
    //连接没有打开返回true
    public abstract boolean isConnectionPending()

*  读取

为了读取SocketChannel，首先需要创建一个ByteBuffer，通道可以在其中存储数据，然后将这个ByteBuffer传给read()方法

public abstract  int read(ByteBuffer dst) throws IOException

通道会尽可能多的数据填充缓存区，然后返回放入的字节数。如果遇到流末尾，通道会用所有剩余的字节数填充缓存区，而且下一次调用read时返回-1。如果通道是阻塞的，这个方法至少读取一个字节，或者返回-1，也可能抛出异常。如果通道是非阻塞的，这个方法可能返回0。有时如果能从一个源填充多个缓存区，这成为散布：

public final long read(ByteBuffer[] dsts) throws IOException
    public final long read(ByteBuffer[] dsts,int offset, int length) throws IOException

ByteBuffer[] buffers=new ByteBuffer[2];
    buffers[0]=ByteBuffer.allocate(1000);
    buffers[1]=ByteBuffer.allocate(1000);
    while(buffers[1].hasRemaining() && channel.read(buffers)!=-1);

*  写入  
    Socket通道提供了读写方法，一般情况下它们是全双工的。想要写入，只需填充一个ByteBuffer，将其回绕，然后传给某个写入方法，这个方法把数据复制到输出时将缓冲区排空，这与读取正好相反。

public abstract int wirte(ByteBuffer src) throws IOException

与读取一样，如果通道是非阻塞的，write方法不能保证会写入缓存区的全部内容。所以要通过循环来实现全部数据的写入。将多个缓存区写入一个Socket通常很有用。这称为聚集。

public final long write(ByteBuffer[] dsts) throws IOException
    public final long write(ByteBuffer[], int offset, int length)

*  关闭  
    就像正常关闭Socket一样，在用完通道后应该将其关闭，释放它可能使用的端口和其他任何资源

public void close()throws IOException

如果不确定通道是否关闭了可以使用下面方法检查

public boolean isOpen()

> 在java 7开始，SocketChannel实现了AutoCloseable，所以可以在try-with-resources中使用

## 2. ServerSocketChannel ##

ServerSocketChannel类只有一个目的：接受入站连接。你无法读取、写入或连接ServerSocketChannel。它唯一支持的操作就是接受一个入站连接，这个类本身只声明了4个方法，其中accept()最重要。ServerSocektChannel还从其超类继承了几个方法，主要与向Selector注册来得到入站连接通知有关。最后，与所有的通道一样，它有一个close()方法，用来关闭服务器Socket

> ServerSocketChannel 用于监听传入的连接请求并创建服务器端的套接字，而 SocketChannel 用于与服务器建立连接并进行数据交互。它们在功能、用途、连接建立、阻塞与非阻塞以及多路复用等方面有所区别

*  创建服务器Socket通道

使用静态方法open创建一个新的ServerSocketChannel对象，但并不打开一个新的服务器Socket，而只是创建这个对象。在使用之前，需要调用Socket方法获取对等端的ServerSocket。此时你可以为该ServerSocket设置各种服务选项。然后将ServerSocket绑定到指定端口的SocketAddress。

ServerSocketChannel server=ServerSocketChannel.open()
    SocketAddress. address=new InetSocketAddress(80)
    server.bind(address)

接受连接

一旦打开并绑定了ServerSocketChannel对象，accept()方法就可以监听连接了：

public abstract SocketChannel accept() throws IOException

accept()可以在阻塞或非阻塞模式操作。在阻塞模式下，accept()方法等待入站连接。然后它接受一个连接，并返回连接到远程客户端的一个SocketChannel对象。在建立连接之前，线程无法进行任何操作。这种策略适用立即响应每一个请求的简单服务器。阻塞模式是默认模式。ServerSocketChannel处于非阻塞模式下连接。在这种情况下，如果没有入站就连接，accept方法会返回null。非阻塞模式适合于需要为每个连接完成大量工作的服务器，这样就可以并行地处理多个请求。非阻塞模式一般于Selector结合使用。为了使ServerSocektChannel处于非阻塞模式，要向configureBlocking()方法传入false。accept方法声明为出现错误时抛出一个IOException异常。IOException的几个子类以及几个运行时异常可以指示更详细的问题：

*  ClosedChannelException：关闭后无法重新打开一个ServerSocketChannel
 *  AsychronousCloseException：执行accept时，另一个线程关闭了这个ServerSocketChannel
 *  ClosedByInterruptException：一个阻塞ServerSocedChannel在等待时，另一个线程中断了这个线程
 *  NotYetBoundException：调用了open()，但在调用accept()之前没有将ServerSocketChannel的对等端ServerSocket地址绑定。这是一个运行时异常
 *  SecurityException：安全管理器拒绝这个应用程序绑定所请求的方法

## 2. Channels类 ##

Channels时一个简单的工具类，可以将传统的基于I/O流、阅读器和书写器包装在通道中，也可以从通道转换为基于I/O的流、阅读器和书写器。如果出于性能考虑将，希望在程序的一部分中使用新I/O模型，但同时仍要与处理流的传统API交换吗，这个类会很有用。它有一些方法可以从流转换为通道，还有一些方法可以从通道转换为流、阅读器、书写器：

public static InputStream newInputStream(ReadableByteChannel ch)
    public static OutputStream newOutputStream(WritableByteChannel ch)
    public static ReadableByteChannel newChannel(InputStream in)
    public static WritableByteChannel newChannel(OutputStream out)
    public static Reader newReader(ReadableByteChannel channel,CharsetDecoder decoder, int mininumBufferCapacity)
    public static Writer newWriter(WritableByteChannel ch, String encoding)

SocketChannel类实现了这些方法签名中出现的ReadableByteChannel和WritableByteChannel接口。ServerSocketChannel二者都没实现，所以不能对ServerSocketChannel进行读/写

## 3. 异步通道（Java 7） ##

在网络编程中，异步通道也被广泛应用于处理非阻塞的I/O操作。Java的异步通道提供了AsynchronousSocketChannel和AsynchronousServerSocketChannel等类，用于实现异步的套接字通信。下面是一个简单的示例，演示了如何使用异步通道进行网络通信：

import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousServerSocketChannel;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.nio.channels.CompletionHandler;
    import java.util.concurrent.ExecutionException;
    
    public class Server {
        
        public static void main(String[] args) {
        
            try {
        
                // 创建异步服务器套接字通道
                AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open();
                serverChannel.bind(new InetSocketAddress(8080));
    
                // 监听连接请求
                serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
        
                    @Override
                    public void completed(AsynchronousSocketChannel clientChannel, Void attachment) {
        
                        // 处理连接成功的回调
                        // 可在这里进行读写操作等处理
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        clientChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
        
                            @Override
                            public void completed(Integer bytesRead, ByteBuffer buffer) {
        
                                // 处理读取数据的回调
                                buffer.flip();
                                byte[] data = new byte[buffer.remaining()];
                                buffer.get(data);
                                String message = new String(data);
                                System.out.println("Received message: " + message);
                            }
    
                            @Override
                            public void failed(Throwable exc, ByteBuffer buffer) {
        
                                // 处理读取数据失败的回调
                                exc.printStackTrace();
                            }
                        });
    
                        // 继续监听连接请求
                        serverChannel.accept(null, this);
                    }
    
                    @Override
                    public void failed(Throwable exc, Void attachment) {
        
                        // 处理连接请求失败的回调
                        exc.printStackTrace();
                    }
                });
    
                // 阻塞主线程，保持服务器运行
                Thread.currentThread().join();
            } catch (IOException | InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }

使用CompletionHandler：  
CompletionHandler接口定义了异步操作完成后的回调方法，它包含两个方法：completed和failed。这些方法分别在操作成功完成和操作失败时被调用。

*  创建并发起异步操作：首先，需要创建一个实现了CompletionHandler接口的回调对象，并重写其中的方法。然后，使用异步操作的方法，如read()、write()等，将回调对象作为参数传递进去。
 *  定义回调逻辑：在回调对象中，根据具体业务需求实现completed和failed方法。completed方法在异步操作成功完成时被调用，可以在该方法中处理操作结果；failed方法在异步操作失败时被调用，可以在该方法中处理异常情况。下面是一个示例代码，展示了如何使用CompletionHandler：

AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    
    channel.read(buffer, null, new CompletionHandler<Integer, Void>() {
        
        @Override
        public void completed(Integer bytesRead, Void attachment) {
        
            // 处理读取操作成功完成的回调
            buffer.flip();
            byte[] data = new byte[buffer.remaining()];
            buffer.get(data);
            String message = new String(data);
            System.out.println("Received message: " + message);
        }
    
        @Override
        public void failed(Throwable exc, Void attachment) {
        
            // 处理读取操作失败的回调
            exc.printStackTrace();
        }
    });

使用Future：  
Future是一个表示异步操作结果的抽象类，它提供了一系列方法来管理和获取操作的结果。Future可以通过异步操作的返回值或异常来表示操作的完成状态。

*  发起异步操作：使用异步操作的方法，如submit()、schedule()等，来发起异步操作。这些方法会返回一个Future对象，可以使用该对象来管理操作。
 *  获取操作结果：通过调用Future的get()方法可以阻塞等待操作的完成，并获取操作的结果。该方法会返回操作的结果，或者在操作完成前阻塞线程。  
    下面是一个示例代码，展示了如何使用Future：

ExecutorService executor = Executors.newSingleThreadExecutor();
    Future<String> future = executor.submit(() -> {
        
        // 执行异步操作，返回结果
        return "Operation completed successfully";
    });
    
    try {
        
        String result = future.get(); // 阻塞等待操作完成，并获取结果
        System.out.println("Operation result: " + result);
    } catch (InterruptedException | ExecutionException e) {
        
        e.printStackTrace();
    }

需要注意的是，使用Future的get()方法是阻塞的，即如果操作没有完成，调用线程会一直等待。可以通过调用isDone()方法来判断操作是否完成，或者使用get(timeout, unit)方法设置超时时间。

## 4. Socket选项 ##

从Java 7，SocketChannel、ServerSocketChannel、AsychronousServerSocketChannel、AsynchronousSocketChannel和DatagramChannel都实现了NetworkChannel接口。这个接口的主要用途是支持各种TCP选项，如TCP\_NODELAY、SO\_TIMEOUT、SO\_LINGER、SO\_SNDBUF、SO\_RCVBUF和SO\_KEEP LIVE。不论是在Socket上设置还是在通道上设置，这些选项在底层TCP栈都有相同的含义。不过，这些选项的接口有些不同。并非对应所支持的各个选项有单独的方法，通道类有3个方法获取、设置和列出所支持的选项：

<T> T getOption(SocketOption<T> name) throws IOException
    <T> NetworkChannel setOption(SocketOption<T> name,T value)throws IOException
    Set<SocketOption<?>> supportOptions()

SocketOption类是一个泛型类，指定了各个选项的名字和类型。类型参数确定这个选项是一个boolean、Integer，还是NetworkInterface。StandarSocketOptions类为Java能识别的11个选项提供了相应的常量。

*  SocketOption<NetworkInterface> StandardSocketOptions.IP\_MULTICAST\_IF
 *  SocketOption<Boolean> StandardSocektOptions.IP\_MULTICAST\_LOOP
 *  SocketOption<Integer> StandardSocektOptions.IP\_MULTICAST\_TTL
 *  SocketOption<Integer> StandardSocektOptions.IP\_TOS
 *  SocketOption<Boolean> StandardSocektOptions.SO\_BROADCAST
 *  SocketOption<Boolean> StandardSocektOptions.SO\_KEEPALIVE
 *  SocketOption<Integer> StandardSocektOptions.SO\_LINGER
 *  SocketOption<Integer> StandardSocektOptions.SO\_RCVBUF
 *  SocketOption<Boolean> StandardSocektOptions.SO\_REUSEADDR
 *  SocketOption<Integer> StandardSocektOptions.SO\_SNDBUF
 *  SocketOption<Boolean> StandardSocektOptions.TCP\_NODELAY

例如，下面的代码段会打开一个客户端网络通道，并设置SO\_LINGER为240s

NetworkChannel channel=SocketChannel.open()
    channel.setOption(StandardSocketOptions.SO_LINGER,240)

不同的通道和Socket支持不同的选项。例如，ServerSocketChannel支持SO\_REUSEADDR和SO\_RCVBUF，但不支持SO\_SNDBUF。如果试图设置通道不支持的一个选项，会抛出 一个UnsupportedOperationException异常。

## 5. 就绪选择 ##

对于网络编程，NIO API得第二个部分是就绪选择，即能够选择读写时不阻塞的Socket。这主要针对于服务器，但对于打开多个窗口并运行多个并发连接客户端来说，也可以利用这个特性。为了完成就绪选择，要将不同的通道注册到一个Selector对象。每个通道分配有一个SelectionKey。然后程序可以询问这个Selector对象，哪些通道已经准备就绪可以无阻塞完成你希望完成的操作，可以请求Selector对象返回相应的键集合。

## 6. Selector类 ##

Selector唯一的构造函数是一个保护类型方法。一般情况下，要调用静态工厂方法Selector.open()来创建新的选择器。

public static Selector open() throws IOException

下一步是向选择器添加通道，Selector类并没有增加通道的方法。register()方法在SelectableChannel类中声明。并不是所有的通道都是可选择的（FileChannel就不可选择），不过所有网络通道都是可选择的。因此，通过将选择器传递给通道的一个注册方法，就可以向选择器注册这个通道：

public final SelectionKey register(Selector sel, int ops) throws ClosedChannelException
    public final SelectionKey register(Selector sel,int ops, Object att)
    throws  ClosedChannelExcption

我觉得这种方法是一种倒退，但使用起来并不难，第一个参数是通道要向哪个选择器注册。第二个参数是SelectionKey类中的一个命名常量，标识通道所注册的操作。SelectionKey类中的一个命名常量，标识通道所注册的操作。SelectionKey定义了4个，用于选择操作类型：

*  SelectionKey.OP\_ACCEPT
 *  SelectionKey.OP\_CONNECT
 *  SelectionKey.OP\_READ
 *  SelectionKey.OP\_WRITE

这些操作都是位标志政型常量（1、2、4等）。因此，如果一个通道需要在同一个选择器中关注多个操作（例如读和写一个Socket），只要在注册时利用位“或”操作符组合这些常量即可

channel.register(selector,SelectionKey.OP_READ| SelectionKey.OP_WRITE)

第二个参数是可选的，这是键的附件。这个对象通常用于存储连接的状态。例如要实现一个WEB服务器，可能要附加一个FileInputStream或FileChannel，这个流或通道连接到服务器提供给客户端的本地文件。不同的通道注册到选择器后，就可以随时查询选择器，找出哪些通道已经准备好进行处理。有3个方法可以选择就绪的通道。它们的区别在于寻找就绪通道等待的时间，第一个selectNow()方法会完成非阻塞选择，如果当前没有准备好要处理的连接，它会立即返回：

public abstract int selectNow() throws IOException

另外两个方法是阻塞的：

public abstract int select() throws IOException
    public abstract int select(long timeout) throws IOException

第一个方法在返回前会等待，直到至少有一个注册的通道准备好可以进行处理，第二个在返回0前只等待不超过timeout毫秒。如果没有通道就绪程序就不做任何操作。当知道有通道已经准备好处理时，就可以使用selectKeys()方法获取就绪通道：

public abstract Set<SelectionKey> selectedKeys()

迭代处理返回的集合时，要依次处理各个SelectionKey。还可以从迭代器中删除键，告诉这个选择器这个键已经得到了处理。否则选择器在以后循环时还会通知你有这个键。最后当关闭服务器或不再需要选择器时，应当将其关闭：

public abstract void close() throws IOException

这个步骤会释放与选择器关联的所有资源。更重要的是，它取消了向选择器注册的所有键，并中断被这个选择器的某个选择方法所阻塞的线程。

## 7. SelectionKey类 ##

SelectionKey对象相当于通道的指针，它们还可以保存一个对象附件，一般会存储这个通道上的连接的状态。将一个通道注册到一个选择器时，register()方法会返回SelectionKey对象，不过通常你不需要保存这个引用，selectedKeys()方法可以在Set中再次返回相同的对象，一个通道可以注册多个选择器。当从所选择的键集合中获取一个SelectionKey时，通常首先要测试这些键能进行哪些操作。有一下四种可能：

public final boolean isAcceptable()
    public final boolean isConnecable()
    public final boolean isReadable()
    public final boolean isWritable()

在有些情况下，选择器只测试一种可能性，也只返回完成这种操作的键。但如果选择器确实要测试多种就绪状态，就要在操作前先测试通道对于哪个操作进入就绪状态，也有可能通道准备好可以完成多个操作。一旦了解了与键关联通道准备好完成何种操作，就可以用channel()方法来获取这个通道：

public abstract SelectableChannel channel()

如果在保存状态信息的SelectionKey存储了一个对象，就可以用attachment()方法获取该对象：

public final Object attachment()

最后如果结束使用连接，就要撤销器SelecitonKey对象的注册，这样选择器就不会浪费资源再去查询它是否准备就绪

public abstract void cancel()

不过，只有在未关闭通道时这个步骤才有必要。如果关闭通道，会自动在所有选择器撤销对应这个通道的所有键的注册。类似地，关闭选择器会使这个选择器中的所有键都失效。