interfaces which it implements): int read(bytebuffer) throws IOException; int read(bytebuffer[] buffers) throws IOException; int read(bytebuffer[] buf

Size: px

Start display at page:

Download "interfaces which it implements): int read(bytebuffer) throws IOException; int read(bytebuffer[] buffers) throws IOException; int read(bytebuffer[] buf"

付费
5 years ago
Views:

1 这章讨论了 JDK 1.4 UDP 数据报可伸缩 I/O 的使用 UDP 的通道在 UDP 之上的可伸缩 I/O 使用我们在之前节所遇到的 DatagramChannel 类执行导入语句下面的 Java 导入语句假设在这章的示例中至始至终存在. import java.io.*; import java.net.*; import java.nio.*; import java.nio.channels.*; import java.util.*; 创建一个 DatagramChannel DatagramChannel 类有和我们之前已经遇到的 SocketChannel 一样的打开和关闭方法 : static DatagramChannel open() throws IOException boolean isopen(); void close() throws IOException; 打开一个 DatagramChannel 返回一个阻塞模式的通道并可准备来使用. 它的套接字可以绑定如果需要的话如节所讨论的. 通道必须被关闭当它结束的时候. 警告 (Caveat): 见节的关于关闭注册的通道的说明, 同样适用于 DatagramChannel 连接操作 DatagramChannel 可以被直接连接和断开, 和通过它的 DatagramSocket 一样 : DatagramChannel connect(socketaddress target) throws IOException; DatagramChannel disconnect() throws IOException; boolean isconnected(); 这些方法操作等同于 DatagramSocket 一致的方法. 不像 SocketChannel 的连接方法,connect 方法不是一个 UDP 套接字连接操作的非阻塞版本 :DatagramChannel.connect 和 DatagramSocket.connect 语义上是相同的. isconnected 方法判断本地套接字是否已经连接上. 如我们在看到的, 连接一个数据报套接字仅仅是一个本地的操作而没有设计网络, 也就是说, 没有什么去阻塞 DatagramChannel I/O DatagramChannel 有我们已经在节看到的读和写操作,(FIXME:as required by the

2 interfaces which it implements): int read(bytebuffer) throws IOException; int read(bytebuffer[] buffers) throws IOException; int read(bytebuffer[] buffers, int offset, int length) throws IOException; int write(bytebuffer) throws IOException; int write(bytebuffer[] buffers)throws IOException; int write(bytebuffer[] buffers, int offset, int length) throws IOException; 现在, 这个 API 完全不同于 (FIXME:the DatagramPacket-orirnted API exported by the DatagramSocket class) 在节和节描述的. 特别是, 没有地方去指定一个输出套接字的目的地址, 或者指定一个传入的数据报的接收的源地址. 因为这个原因,read 和 write 方法在 DatagramChannel 受限于一个重要的语义约束 : 如果关联的套接字连接了如节定义的, 它们才能被使用. 在这种情况下, 一个接收或者发送的数据包的源或者目标地址只能是连接的套接字, 所以上面的读 / 写 API 是适合的. 为了处理未连接的数据报套接字,DatagramChannel 提供了两个新的方法 : SocketAddress receive(bytebuffer buffer) throws IOException; int send(bytebuffer buffer, SocketAddress target) throws IOException; 这些分别对应于 DatagramSocket.receive 和 DatagramSocket.send 方法.DatagramChannel.send 的 target 参数是传输的远程目标.DatagramChannel.receive 的返回值为 SocketAddress, 如我们在 2.2 节看到的, 代表了传输的远程源阻塞的 UDP I/O 如我们在 10.1 节看到的, 一个 DatagramChannel 以阻塞模式创建. 在这种模式下 : (a) 一个读操作阻塞直到一个传入的数据报已经接收到套接字接收缓冲区, 如果没有任何东西存在. (b) 一个写操作阻塞直到空间可用使在套接字发送缓冲区中的输出数据报排队, 也就是说, 直到有足够的数据报预先传输到网络上排队 : 这将延迟, 如果发生的话, 它通常非常短暂. 由于数据报以最大的速录传输非阻塞的 UDP I/O 一个 DatagramChannel 可被放置于非阻塞模式 : SelectableChannel configureblocking(boolean block) throws IOException; boolean isblocking(); 在非阻塞模式中,write 和 send 方法可以返回 0, 表明套接字发送缓冲区已经满了, 没有传输可以被执行 ; 类似的,read 和 receive 方法可以分别返回 0 或者 null, 表明没有数据报可用. 一个简单的非阻塞的 UDP I/O 顺序再示例 10.1 中说明. ByteBuffer buffer = ByteBuffer.allocate(8192); DatagramChannel channel= DatagramChannel.open(); channel.configureblocking(false); SocketAddress address

3 = new InetSocketAddress( localhost, 7); buffer.put( ); while (channel.send(buffer, address) == 0) ; // do something useful while ((address = channel.receive(buffer)) == null) ; // do something useful Example 10.1 简单的非阻塞的 UDP 客户端 I/O 如注释所说, 程序应该做一些有用的工作或者休眠而不是无知的自旋 (FIXME:spinning mindlessly) 当 I/O 传输返回 0 的时候多路复用 UDP 中的可伸缩中的可伸缩 I/O 操作 UDP 中的可伸缩 I/O 操作都适用于 DatagramChannel 对象. 它们的 ' 准备 ' 状态的含义在表 10.1 展示. 操作含义表 10.1 UDP 中的可伸缩 I/O 操作 OP_READ 数据存在于套接字接收缓冲区中或者异常挂起. 在套接字发送缓冲区中存在空间或者异常挂起. 在 UDP 中,OP_WRITE 几乎总是准备好除了在套接字发送缓冲区中没有空间可用的期间 OP_WRITE. 最后只注册 OP_WRITE 一旦这个缓冲区满的条件被探测到, 也就是说, 当一个通道写返回少于请求的写的长度, 注销 OP_WRITE 一旦它被清理了, 也就是说, 一个通道写完全成功了多路复用 I/O 一个 DatagramChannel 可以使用一个选择器注册, 调用 Selector.select 方法去等待通道的可读或者可写, 如节讨论的 : DatagramChannelchannel = DatagramChannel.open(); // bind to port 1100 channel.bind(new InetSocketAddress(1100)); Selector selector = Selector.open(); // register for OP_READ channel.register(selector, SelectionKey.OP_READ); // Select selector.select(); 示例一个简单的多路复用 UDP echo 服务器在示例 10.2 中展示 ( 书中写的是 List<Datagram>, 正确的应该是 List<DatagramPacket>). public class NIOUDPEchoServer implements Runnable { static final int TIMEOUT = 5000;// 5s private ByteBuffer buffer = ByteBuffer.allocate(8192); private DatagramChannel channel; private List<DatagramPacket> outputqueue = new LinkedList<DatagramPacket>();

4 // Create new NIOUDPEchoServer public NIOUDPEchoServer(int port) throws IOException { this.channel = DatagramChannel.open(); channel.socket().bind(new InetSocketAddress(port)); channel.configureblocking(false); // Runnable. run public void run() { Selector selector = Selector.open(); channel.register(selector, SelectionKey.OP_READ); // loop while there are any registered channels while (!selector.keys().isempty()) { int keysadded = selector.select(timeout); // Standard post-select processing Set selectedkeys = selector.selectedkeys(); synchronized (selectedkeys) { Iterator it = selectedkeys.iterator(); while (it.hasnext()) { SelectionKey key = (SelectionKey) it.next(); it.remove(); if (!key.isvalid()) continue; if (key.isreadable()) handlereadable(key); if (key.iswritable()) handlewritable(key); // while // synchronized // while // try catch (IOException e) { // handle readable key void handlereadable(selectionkey key) { DatagramChannel channel = (DatagramChannel) key.channel(); SocketAddress address = channel.receive(buffer); if (address == null) return; // no data channel.send(buffer, address); int count = buffer.remaining();

5 if (count > 0) { // Write failure: queue the write request // as a DatagramPacket, as this nicely holds // the data and reply address byte[] bytes = new byte[count]; buffer.get(bytes); outputqueue.add(new DatagramPacket(bytes, count, address)); // Register for OP_WRITE key.interestops(selectionkey.op_read SelectionKey.OP_WRITE); catch (IOException e) { // handlereadable() // handle writable key void handlewritable(selectionkey key) { DatagramChannel channel = (DatagramChannel) key.channel(); while (!outputqueue.isempty()) { DatagramPacket packet = outputqueue.get(0); buffer.put(packet.getdata()); channel.send(buffer, packet.getsocketaddress()); if (buffer.hasremaining()) // write failed, retry return; outputqueue.remove(0); // All writes succeeded & queue empty, so // deregister for OP_WRITE key.interestops(selectionkey.op_read); catch (IOException e) { // handlewritable() // end of NIOUDPEchoServer 这个服务器永远不会在 I/O 操作中阻塞, 只有在 Selector.select 中. 如我们在节看到的, 它比 TCP 服务器简单点, 不必去管理客户端连接. 在多路复用 I/O, 这个意味着它不必去管理 SelectionKey.isAcceptable 的情况, 或者处理其它的套接字的连接. 它需要去管理输出队列包含了数据和目标地址. 幸运地是,java.net.DatagramPacket 类恰好包含了这些数据项目已经被传递.

JavaIO.PDF

JavaIO.PDF O u t p u t S t ream j a v a. i o. O u t p u t S t r e a m w r i t e () f l u s h () c l o s e () public abstract void write(int b) throws IOException public void write(byte[] data) throws IOException