Java面试15|网络

netstat -n | awk ‘/^tcp/ {++S[$NF]} END{for(a in S) print a, S[a]}‘  

public class MultiplexerTimeServer implements Runnable {

	private Selector selector;

	private ServerSocketChannel servChannel;

	private volatile boolean stop;

	/**
	 * 初始化多路复用器、绑定监听端口
	 */
	public MultiplexerTimeServer(int port) {
		try {
			selector = Selector.open();
			
			// Channel主要用来读写网络上的数据的。打开ServerSocketChannel,用于监听客户端的连接，它是所有客户端连接的父管道
			servChannel = ServerSocketChannel.open();
			
			// 设置为非阻塞模式
			servChannel.configureBlocking(false);
			
			// 绑定监听端口 8080
			servChannel.socket().bind(new InetSocketAddress(port), 1024);
			
			/*
			 * Selector会不断地轮询在其上的Channel，如果某个Channel上面有新的TCP
			 * 连接接入、读和写事件，这个Channel就处于就绪状态
			 * 
			 * 注册到Reacotr线程的多路复用器Selector上，监听ACCEPT事件
			 */
			servChannel.register(selector, SelectionKey.OP_ACCEPT);
			
			System.out.println("The time server is start in port : " + port);
		} catch (IOException e) {
			e.printStackTrace();
			System.exit(1);
		}
	}

	public void stop() {
		this.stop = true;
	}

	public void run() {
		while (!stop) {
			try {
				// This method performs a blocking selection operation. 
				// It returns only after at least one channel is selected,（只有在至少有一个事件就绪后才会进行返回，所以是阻塞的） 
				// this selector‘s wakeup method is invoked, the current thread is interrupted, 
				// or the given timeout period expires, whichever comes first. 
				selector.select(1000); // 阻塞等待，休眠时间为1s
				Set<SelectionKey> selectedKeys = selector.selectedKeys();
				/*
				 * 当有处于就绪状态的Channel时，selector将返回就绪状态的Channel的SelectionKey
				 * 集合，通过对就绪状态的Channel集合进行迭代，可以进行网络的异步读写操作
				 */
				Iterator<SelectionKey> it = selectedKeys.iterator();
				SelectionKey key = null;
				while (it.hasNext()) {
					key = it.next();
					it.remove();
					try {
						// 事件分发器，单线程选择就绪的事件。
						// I/O处理器，包括connect、read、write等，这种纯CPU操作，一般开启CPU核心个线程就可以。
						// 业务线程，在处理完I/O后，业务一般还会有自己的业务逻辑，有的还会有其他的阻塞I/O，如DB操作，RPC等。只要有阻塞，就需要单独的线程。
						handleInput(key); // 所以在这里其实最好是用其它的线程来处理，而不要影响了事件分发器线程
					} catch (Exception e) {
						if (key != null) {
							key.cancel();
							if (key.channel() != null)
								key.channel().close();
						}
					}
				}
			} catch (Throwable t) {
				t.printStackTrace();
			}
		}

		// 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
		if (selector != null)
			try {
				selector.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
	}

	private void handleInput(SelectionKey key) throws IOException {

		if (key.isValid()) {
			
			// 处理新接入的请求消息
			if (key.isAcceptable()) {
				// 接受一个新的客户端接入请求
				ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
				SocketChannel sc = ssc.accept();
				// 设置客户端为异步非阻塞
				sc.configureBlocking(false); 
				// Add the new connection to the selector
				sc.register(selector, SelectionKey.OP_READ);
			}
			if (key.isReadable()) {
				/*
				 * 读取数据
				 * 读取到的字节数，返回值有以下有三种结果：
				 * （1）大于0，读取到字节，对其进行解编码
				 * （2）等于0，没有读取到字节，南纺股份正常场景，忽略
				 * （3）-1 ，链路已经关闭，需要关闭SocketChannel，释放资源
				 */
				SocketChannel sc = (SocketChannel) key.channel();
				ByteBuffer readBuffer = ByteBuffer.allocate(1024);
				// 由于设置了SocketChannel为异步非阻塞的，所以它的read是非阻塞的
				int readBytes = sc.read(readBuffer);
				if (readBytes > 0) {
					/*
					 * 将缓冲区当前的limit设置为position,position为0，用于后续对缓冲区的读取操作。
					 * 然后根据缓冲区可读的字节个数创建字节数组，调用get()操作将缓冲区可读的字节数
					 * 组复制到新创建的字节数组中
					 */
					readBuffer.flip();
					byte[] bytes = new byte[readBuffer.remaining()];
					readBuffer.get(bytes);
					String body = new String(bytes, "UTF-8");
					System.out.println("The time server receive order : " + body);
					String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? 
							new java.util.Date(System.currentTimeMillis()).toString() : "BAD ORDER";
					doWrite(sc, currentTime);
				} else if (readBytes < 0) {
					// 对端链路关闭
					key.cancel();
					sc.close();
				} else{
					; // 读到0字节，忽略
				}
					
			}
		}
	}

	private void doWrite(SocketChannel channel, String response)throws IOException {
		/*
		 * 由于SocketChannel是异步非阻塞的，并不能保证一次能够把所有需要发送的数据发送，此时会出现写半包问题。
		 * 需要注册写操作???，不断轮询Selector将没有发送完的bytebuffer发送完毕。可以通过byteBuffer的hasRemain()
		 * 方法判断是否发送完毕。
		 */
		if (response != null && response.trim().length() > 0) {
			// 将应答消息异步发送给客户端
			byte[] bytes = response.getBytes();
			ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
			// 将缓冲区中的字节数据发送
			writeBuffer.put(bytes);
			writeBuffer.flip(); // 缓存区复位
			channel.write(writeBuffer);
		}
	}
}