AsynchronousServerSocketChannel和AsynchronousSocketChannel类的使用

秒速五厘米 2021-08-30 18:52 141阅读 0赞

### AsynchronousServerSocketChannel和AsynchronousSocketChannel类的使用 ###

`AsynchronousServerSocketChannel类是面向流的侦听套接字的异步通道。`1个AsynchronousServerSocketChannel通道是通过调用此类的`open()`方法创建的。新创建的AsynchronousServerSocketChannel已打开但尚未绑定。它可以绑定到本地地址，并通过调用bind()方法来配置为侦听连接。一旦绑定，accept()方法被用来启动接受连接到通道的Socket。尝试在未绑定通道上调用accept()方法将导致引发NotYetBoundException异常。

此类型的通道是线程安全的，可由多个并发线程使用，但在大多数情况下，在任何时候都可以完成一个accept操作。如果线程在上一个接受操作完成之前启动接受操作，则会引发AcceptPendingException异常。

可以使用setOption() 方法设置如下的Socket Option：

<table> 
 <thead> 
  <tr> 
   <th align="left">Option Name</th> 
   <th align="left">Description</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">SO_RCVBUF</td> 
   <td align="left">The size of the socket reveive buffer</td> 
  </tr> 
  <tr> 
   <td align="left">SO_REUSEADDR</td> 
   <td align="left">Re-use address</td> 
  </tr> 
 </tbody> 
</table>

其他的Socket Option是否支持取决于实现。

public static void main(String[] args) throws Exception{ 
            final AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(8888));
            serverSocketChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() { 
    
                @Override
                public void completed(AsynchronousSocketChannel result, Object attachment) { 
                    serverSocketChannel.accept(null, this);
                    //方法handle用来处理这个链接
                    handle(ch);
                }
    
                @Override
                public void failed(Throwable exc, Object attachment) { 
    
                }
            });
        }

AsynchronousServerSocketChannel类的结构信息：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]  
AsynchronousServerSocketChannel类的继承关系：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

因为AsynchronousServerSocketChannel类是抽象类，所以不能直接new实例化，需要借助于open()方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

AsynchronousSocketChannel类是面向流的连接套接字的异步通道:

使用AsynchronousSocketChannel 类的open() 方法创建的是未连接状态的Asynchronous-SocketChannel对象，之后再使用connect() 方法将未连接的AsynchronousSocketChannel变成已连接的AsynchronousSocketChannel对象，详述如下:

1.  创建AsynchronousSocketChannel是通过调用此类定义的open()方法，新创建的AsynchronousSocketChannel呈已打开但尚未连接的状态。当连接到AsynchronousServerSocketChannel的套接字时，将创建连接的AsynchronousSocketChannel对象。不可能为任意的、预先存在的Socket创建异步套接字通道。
2.  通过调用connect()方法将未连接的通道变成已连接，连接后该通道保持连接，直到关闭。是否连接套接字通道可以通过调用其getRemoteAddress()方法来确定。尝试在未连接的通道上调用I0操作将导致引发NotYetConnectedException异常。

此类型的通道可以安全地由多个并发线程使用。它们支持并发读写，虽然最多一次读取操作，并且一个写操作可以在任何时候未完成。如果一个线程在上一个读操作完成之前启动了read 操作，则会引发ReadPendingException异常。类似的，尝试在前一个写操作完成之前启动一个写运算将会引发一个WritePendingException异常。

可以使用setOption()方法设置如下的Socket Option：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]  
类结构信息：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

此类定义的read()和write()方法允许在启动读或写操作时指定超时。如果在操作完成之前超时，则操作将以InterruptedByTimeoutException异常完成。超时可能会使通道或基础连接处于不一致状态。如果实现不能保证字节没有从通道中读取，那么它就会将通道置于实现特定的错误状态，随后尝试启动读取操作会导致引发未指定的运行时异常。类似的，如果写操作超时并且实现不能保证字节尚未写人信道，则进一步尝试写入信道会导致引发未指定的运行时异常。如果超时时间已过，则不定义I/O操作的缓冲区或缓冲区序列的状态，应丢弃缓冲区，或者至少要注意确保在通道保持打开状态时不访问缓冲区。所有接受超时参数的方法都将值处理得小于或等于零，这意味着I/O操作不会超时。

AsynchronousSocketChannel类的继承关系：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]

#### 接受方式1 ####

服务端：

public static void main(String[] args) throws Exception{ 
            final AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open()
                    .bind(new InetSocketAddress(8088));
            serverSocketChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() { 
                @Override
                public void completed(AsynchronousSocketChannel result, Object attachment) { 
                    try { 
                        serverSocketChannel.accept(null, this);
                        System.out.println("public void completed ThreadName=" + Thread.currentThread().getName());
                        ByteBuffer byteBuffer = ByteBuffer.allocate(20);
                        Future<Integer> future = result.read(byteBuffer);
                        System.out.println(new String(byteBuffer.array(), 0, future.get()));
                        result.close();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } catch (ExecutionException e) { 
                        e.printStackTrace();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
    
                @Override
                public void failed(Throwable exc, Object attachment) { 
                    System.out.println("public void failed");
                }
            });
            while (true){ }
        }

客户端A：

public static void main(String[] args) throws Exception{ 
            Socket socket = new Socket("localhost", 8088);
            OutputStream outputStream = socket.getOutputStream();
            outputStream.write("我来自客户端A".getBytes());
            outputStream.flush();
            outputStream.close();
        }

客户端B：

public static void main(String[] args) throws Exception{ 
            AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();
            socketChannel.connect(new InetSocketAddress("localhost", 8088), null,
                    new CompletionHandler<Void, Object>() { 
                        @Override
                        public void completed(Void result, Object attachment) { 
                            try { 
                                Future<Integer> future = socketChannel.write(ByteBuffer.wrap("我来自客户端B".getBytes()));
                                System.out.println("写入大小：" + future.get());
                                socketChannel.close();
                            } catch (InterruptedException e) { 
                                e.printStackTrace();
                            } catch (ExecutionException e) { 
                                e.printStackTrace();
                            } catch (IOException e) { 
                                e.printStackTrace();
                            }
                        }
    
                        @Override
                        public void failed(Throwable exc, Object attachment) { 
    
                        }
                    });
            Thread.sleep(1000);
        }

先运行服务端再运行客户端A：

public void completed ThreadName=Thread-13
    我来自客户端A

再运行客户端B：

public void completed ThreadName=Thread-13
    我来自客户端A
    public void completed ThreadName=Thread-1
    我来自客户端B

#### 接受方式2 ####

服务端：

public static void main(String[] args) throws Exception{ 
            AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open()
                    .bind(new InetSocketAddress(8088));
            System.out.println("A " + System.currentTimeMillis());
            Future<AsynchronousSocketChannel> future = serverSocketChannel.accept();
            System.out.println("B " + System.currentTimeMillis());
            AsynchronousSocketChannel socketChannel = future.get();
            System.out.println("C " + System.currentTimeMillis());
            ByteBuffer byteBuffer = ByteBuffer.allocate(20);
            System.out.println("D " + System.currentTimeMillis());
            Future<Integer> readFuture = socketChannel.read(byteBuffer);
            System.out.println("E " + System.currentTimeMillis());
            System.out.println(new String(byteBuffer.array(), 0, readFuture.get()));
            System.out.println("F " + System.currentTimeMillis());
            Thread.sleep(40000);
        }

运行服务端，再运行客户端A：

A 1591859108766
    B 1591859108767
    C 1591859114060
    D 1591859114060
    E 1591859114062
    我来自客户端A
    F 1591859114062

运行服务端，在运行客户端B：

A 1591859210328
    B 1591859210330
    C 1591859213775
    D 1591859213775
    E 1591859213778
    我来自客户端B
    F 1591859213779

#### 重复读与重复写出现异常 ####

服务端：

public static void main(String[] args) throws Exception{ 
            AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open()
                    .bind(new InetSocketAddress(8088));
            Future<AsynchronousSocketChannel> socketChannelFuture = serverSocketChannel.accept();
            AsynchronousSocketChannel socketChannel = socketChannelFuture.get();
            ByteBuffer byteBuffer = ByteBuffer.allocate(20);
            Future<Integer> future1 = socketChannel.read(byteBuffer);
            Future<Integer> future2 = socketChannel.read(byteBuffer);
        }

先运行服务端，再运行客户端B：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]

因为read()方法是非阻塞的，所以执行第1个read()方法后立即继续执行第2个read()方法，但由于第1个read()方法并没有完成读的操作，因为并没有调用future.get()方法，因此出现ReadPendingException异常。  
重复写也是同样的道理，出现WritePendingException异常，在此不再举例说明。

### 读数据 ###

`public abstract <A> void read(ByteBuffer dst,long timeout,TimeUnit unit,A attachment,CompletionHandler<Integer,? super A> handler)`方法的作用是将此通道中的字节序列读入给定的缓冲区。此方法启动一个异步读取操作，以便将该通道中的字节序列读人给定的缓冲区。

*  handler参数是在读取操作完成或失败时调用的CompletionHandler。传递给completed()方法的结果是读取的字节数，如果无法读取字节，则为-1，因为信道已达到end-of-stream。
 *  如果指定了timeout 并且在操作完成之前发生超时的情况，则操作将以异常InterruptedByTimeoutException完成。在发生超时的情况下，实现无法保证字节没有被读取，或者不会从通道读取到给定的缓冲区，那么进一步尝试从通道读取将导致引发不运行时异常，否则，此方法的工作方式与`public final <A> void read(ByteBuffer dst, A attachment,CompletionHandler<Integer,? super A> handler)方法相同。`

public static void main(String[] args) throws Exception{ 
            final AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open()
                    .bind(new InetSocketAddress(8088));
            serverSocketChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() { 
                @Override
                public void completed(AsynchronousSocketChannel ch, Object attachment) { 
                    serverSocketChannel.accept(null, this);//continue next accept
                    ByteBuffer byteBuffer = ByteBuffer.allocate(Integer.MAX_VALUE / 10000);
                    ch.read(byteBuffer, 20, TimeUnit.SECONDS, null, new CompletionHandler<Integer, Object>() { 
                        @Override
                        public void completed(Integer result, Object attachment) { 
                            if (result == -1)
                                System.out.println("客户端没有传输数据就执行close了，到stream end");
                            if (result == byteBuffer.limit())
                                System.out.println("服务端获得客户端完整数据");
                            try { 
                                ch.close();
                                System.out.println("服务端close");
                            } catch (IOException e) { 
                                e.printStackTrace();
                            }
                        }
    
                        @Override
                        public void failed(Throwable exc, Object attachment) { 
                            System.out.println("read public void failed(Throwable exc, Object attachment)");
                            System.out.println("exc getMessage()=" + exc.getClass().getName());
                        }
                    });
                }
    
                @Override
                public void failed(Throwable exc, Object attachment) { 
                    System.out.println("accept public void failed");
                }
            });
            while (true){ 
            }
        }

public static void main(String[] args) throws IOException, InterruptedException { 
            final AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();
            socketChannel.connect(new InetSocketAddress("localhost", 8088), null,
                new CompletionHandler<Void, Object>() { 
                    @Override
                    public void completed(Void result, Object attachment) { 
                        try { 
                            ByteBuffer byteBuffer = ByteBuffer.allocate(Integer.MAX_VALUE / 10000);
                            for (int i = 0; i < Integer.MAX_VALUE / 10000 - 3; i++) { 
                                byteBuffer.put("1".getBytes());
                            }
                            byteBuffer.put("end".getBytes());
                            byteBuffer.flip();
                            int writeSum = 0;
                            //由于write()方法是异步的，所以执行write()方法后
                            //并不能100%将数据写出，所以得通过writeLength变量
                            //来判断具体写出多少字节的数据
                            while (writeSum < byteBuffer.limit()) { 
                                Future<Integer> writeFuture = socketChannel.write(byteBuffer);
                                Integer writeLength = writeFuture.get();
                                writeSum = writeSum + writeLength;
                                System.out.println(writeSum);
                            }
                            socketChannel.close();
                        }catch (InterruptedException e) { 
                            e.printStackTrace();
                        } catch (IOException e) { 
                            e.printStackTrace();
                        } catch (ExecutionException e) { 
                            e.printStackTrace();
                        }
                    }
    
                    @Override
                    public void failed(Throwable exc, Object attachment) { 
                        System.out.println("connect public void failed(Throwable exc, Object attachment)");
                        System.out.println("exc getMessage()=" + exc.getClass().getName());
                    }
                });
            Thread.sleep(20000);
        }

![在这里插入图片描述][20200611163515837.png]

#### 写数据 ####

`public abstract <A> void write(ByteBuffer src,long timeout,TimeUnit unit,A attachment,Co-mpletionHandler<Integer,? super A> handler)`方法的作用是从给定缓冲区向此通道写入一个字节序列。此方法启动异步写人操作，以便从给定缓冲区向此通道写入一个字节序列。

*  handler参数是在写操作完成或失败时调用的CompletionHandler。 传递给completed()方法的结果是写入的字节数。
 *  如果指定了timeout，并且在操作完成之前发生了超时，则它将以异常InterruptedByTimeoutException完成。如果发生超时，并且实现无法保证字节尚未写人或不会从给定的缓冲区写人通道，则进一步尝 试写人信道将导致引发不运行时异常，否则，此方法的工作方式与`public final <A> void write(ByteBuffer src,A attachment,CompletionHandler<Integer,?super A> handler)方法相同。`

public static void main(String[] args) throws IOException { 
            final AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open()
                    .bind(new InetSocketAddress(8088));
            serverSocketChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() { 
                @Override
                public void completed(AsynchronousSocketChannel ch, Object attachment) { 
                    serverSocketChannel.accept(null, this);//继续下一个accept接作
                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                    ch.read(byteBuffer, 10, TimeUnit.SECONDS, null, new CompletionHandler<Integer, Object>() { 
                        @Override
                        public void completed(Integer result, Object attachment) { 
                            if (result == -1) { 
                                System.out.println("客户端没有传输数据就执行close了，到stream end");
                            }
                            if (result == byteBuffer.limit()) { 
                                System.out.println("服务端获得客户端完整数据");
                            }
                            try { 
                                ch.close();
                                System.out.println("服务端close");
                            } catch (IOException e) { 
                                e.printStackTrace();
                            }
                        }
    
                        @Override
                        public void failed(Throwable exc, Object attachment) { 
                            System.out.println("read public void failed(Throwable exc, Object attachment)");
                            System.out.println("exc getMessage()=" + exc.getClass().getName());
                        }
                    });
                }
    
                @Override
                public void failed(Throwable exc, Object attachment) { 
                    System.out.println("accept public void failed");
                }
            });
            while (true){ 
    
            }
        }

public static void main(String[] args) throws IOException, InterruptedException { 
            final AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();
            socketChannel.connect(new InetSocketAddress("localhost", 8088), null,
                    new CompletionHandler<Void, Object>() { 
                        @Override
                        public void completed(Void result, Object attachment) { 
                            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                            for (int i = 0; i < 1000; i++) { 
                                byteBuffer.put("1".getBytes());
                            }
                            byteBuffer.put("end".getBytes());
                            byteBuffer.flip();
                            socketChannel.write(byteBuffer, 1, TimeUnit.SECONDS, null,
                                    new CompletionHandler<Integer, Object>() { 
                                        @Override
                                        public void completed(Integer result, Object attachment) { 
                                            try { 
                                                socketChannel.close();
                                                System.out.println("client close");
                                            } catch (IOException e) { 
                                                e.printStackTrace();
                                            }
                                        }
    
                                        @Override
                                        public void failed(Throwable exc, Object attachment) { 
                                            System.out.println("write public void failed(Throwable exc, Object attachment)");
                                            System.out.println("exc getMessage()=" + exc.getClass());
                                        }
                                    });
                        }
    
                        @Override
                        public void failed(Throwable exc, Object attachment) { 
                            System.out.println("connect public void failed(Throwable exc, Object attachment)");
                            System.out.println("exc getMessage()=" + exc.getClass().getName());
                        }
                    });
            Thread.sleep(5000);
        }

还有另外两个方法:

1.  `read(dsts, offset, length, timeout, unit, attachment, handler);`
2.  `ch. write(srcs, offset, length, timeout, unit, attachment, handler);`

功能就是分散读、聚合写。这两个方法与以下两个方法功能非常相似，在此不再重复演示。

1.  `ch.read(dst, timeout, unit, attachment, handler);`
2.  `ch.write(src, timeout, unit, attachment, handler);`

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/f8854b9d63a7404496f8d00eef3fa406.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/01e18fb75b6948d292fc60df8067d272.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/cec07b08f35f4a5585cb1354720b26cf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/a3751fb16dad459198b0bbc87341ae1c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/2c2f3f5c46244a7ea48ca15a930a27ee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/f32db4a7377b4e36a0ecea573980d02e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/b82b3cf8ecb6474597df5a7be8b1dd3b.png
[20200611163515837.png]: /images/20210813/c44559990bbb4a9c8ab5328531394cc3.png