NIO服务器示例（使用 select( )来为多个通道提供服务）

一时失言乱红尘 2023-02-21 04:55 6阅读 0赞

package com.nio.demo;
    
    import java.net.InetSocketAddress;
    import java.net.ServerSocket;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectableChannel;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    
    public class SelectSockets {
    	public static int PORT_NUMBER = 1234;
    
    	public static void main(String[] args) throws Exception {
    		new SelectSockets().go(args);
    
    	}
    
    	public void go(String[] argv) throws Exception {
    		int port = PORT_NUMBER;
    		if (argv.length > 0) { // Override default listen port
    			port = Integer.parseInt(argv[0]);
    		}
    		System.out.println("Listening on port " + port);
    		// 打开一个 socket channel。注：通道不能复用，一个通道只能对应一个连接，这里的ServerSocketChannel只是通道的监听类。
    		ServerSocketChannel serverChannel = ServerSocketChannel.open();
    		// 获取绑定在Channel上的socket
    		ServerSocket serverSocket = serverChannel.socket();
    		// 创建一个选择器
    		Selector selector = Selector.open();
    		// 监听端口
    		serverSocket.bind(new InetSocketAddress(port));
    		// 设置为非阻塞模式
    		serverChannel.configureBlocking(false);
    		// 将serverChannel注册到选择器上，需要监听的操作为OP_ACCEPT
    		serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    		while (true) {
    			//返回选择器上就绪的通道数量。注：每个通道可能会注册多个操作，任一操作就绪：n>0
    			int n = selector.select();
    			if (n == 0) {
    				continue; // nothing to do
    			}
    			// Get an iterator over the set of selected keys
    			Iterator it = selector.selectedKeys().iterator();
    			// Look at each key in the selected set
    			while (it.hasNext()) {
    				SelectionKey key = (SelectionKey) it.next();
    				// 如果就绪的通道为serverChannel（因为只有serverChannel注册了ACCEPT操作），即意味着有新连接到来
    				if (key.isAcceptable()) {
    					ServerSocketChannel server = (ServerSocketChannel) key.channel();
    					//创建一个对应新连接的通道。注：因为已经通过选择器的检查，accept方法的会立即返回，并且不会为null
    					SocketChannel channel = server.accept();
    					//将此通道注册到选择器上
    					registerChannel(selector, channel, SelectionKey.OP_READ);
    					//模拟客户端向通道发送消息
    					sayHello(channel);
    				}
    				// 如果就绪的通道为连接通道
    				if (key.isReadable()) {
    					//读取数据
    					readDataFromSocket(key);
    				}
    				// 从就绪集合中移除
    				it.remove();
    			}
    		}
    	}
    
    	protected void registerChannel(Selector selector, SelectableChannel channel, int ops) throws Exception {
    		if (channel == null) {
    			return; // could happen
    		}
    		// Set the new channel nonblocking
    		channel.configureBlocking(false);
    		// Register it with the selector
    		channel.register(selector, ops);
    	}
    
    	// ----------------------------------------------------------
    	// Use the same byte buffer for all channels. A single thread is
    	// servicing all the channels, so no danger of concurrent acccess.
    	private ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
    	protected void readDataFromSocket(SelectionKey key) throws Exception {
    		SocketChannel socketChannel = (SocketChannel) key.channel();
    		int count;
    		buffer.clear(); // Empty buffer
    // socketChannel是以非阻塞模式运行，count可能会为0；但因为使用了选择器，第一次读取肯定不会为0。
    		while ((count = socketChannel.read(buffer)) > 0) {
    			buffer.flip(); 
    			while (buffer.hasRemaining()) {
    				socketChannel.write(buffer);
    			}
    			buffer.clear(); // Empty buffer
    		}
    		if (count < 0) {
    // Close channel on EOF, invalidates the key
    			socketChannel.close();
    		}
    	}
    
    // ----------------------------------------------------------
    	/**
    	 * Spew a greeting to the incoming client connection.
    	 *
    	 * @param channel The newly connected SocketChannel to say hello to.
    	 */
    	private void sayHello(SocketChannel channel) throws Exception {
    		buffer.clear();
    		buffer.put("Hi there!\r\n".getBytes());
    		buffer.flip();
    		channel.write(buffer);
    	}
    
    }