06、Netty 基础 之 NIO快速入门

一、案例

1、 编写一个NIO入门案例,实现服务器端和客户端之间的数据简单通讯(非阻塞);
2、 目的:理解NIO非阻塞网络编程机制;
3、 代码;

NIOServer.java

package netty.niostart;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class NIOServer {
	public static void main(String[] args) throws Exception {
		//创建ServerSocketChannel -> 类似于ServerSocket
		ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
		
		//得到一个Selector对象
		Selector selector = Selector.open();
		
		//绑定一个端口6666,在服务器端监听
		serverSocketChannel.socket().bind(new InetSocketAddress(6666));
		//设置为非阻塞
		serverSocketChannel.configureBlocking(false);
		
		//把serverSocketChannel注册到selector,关心事件为OP_ACCEPT
		serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
		
		//循环等待客户端连接
		while(true) {
			if (selector.select(5*1000) == 0) {
				//等待5秒钟,如果没有事件发生,返回
				System.out.println("服务器等待了5秒,无连接");
				continue;
			}
			
			//如果返回的>0,就获取到相关的selectionKey集合
			//1.如果返回的>0,表示已经获取到关注的事件
			//2.selector.selectedKeys()返回关注的集合
			//3.通过selectionKeys反向获取通道
			Set<SelectionKey> selectionKeys = selector.selectedKeys();
			
			//遍历集合
			Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
			
			while (keyIterator.hasNext()) {
				//获取到SelectionKey
				SelectionKey key = keyIterator.next();
				
				//根据key,对应的通道发生的事件,做相应的处理
				if (key.isAcceptable()) {
					//如果是OP_ACCEPT,有新的客户端连接
					//给该客户端生成一个SocketChannel
					SocketChannel socketChannel = serverSocketChannel.accept();
					//将socketChannel设置为非阻塞模式
					socketChannel.configureBlocking(false);
					//将客户端的socketChannel也注册到selector,关注事件为SelectionKey.OP_READ
					//同时给该socketChannel关联一个buffer
					socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(10240));
					
					System.out.println("from " + socketChannel.hashCode() + " 客户端建立了连接");
					
				} else if (key.isReadable()) {
					//发生OP_READ
					//通过key反向获取到对应的channel
					SocketChannel socketChannel = (SocketChannel)key.channel();
					//获取到该channel关联的buffer
					ByteBuffer byteBuffer = (ByteBuffer)key.attachment();
					
					byteBuffer.clear();
					//把当前通道的数据读到buffer中
					try {
						if (socketChannel.read(byteBuffer) == -1) {
							System.out.println("from " + socketChannel.hashCode() + " 客户端断开了连接");
							keyIterator.remove();
							socketChannel.close();
							continue;
						}
						
						//解析客户端数据
						//socket通讯格式可以自己定义:4字节报文长度+报文体
						String request = new String(byteBuffer.array(), 0, byteBuffer.position(), "utf-8");
						System.out.println("from " + socketChannel.hashCode() + " 客户端:" + request);
						
						
						//----------------------------------------------------------------//
						/**
						 * 业务模块处理(能否做成异步,传入socketChannel对象)主线程只负责网络IO读写
						 */
						
						//业务处理
						System.out.println("业务处理...");
						
						//使用新buffer返回
						ByteBuffer newByteBuffer = ByteBuffer.allocate(10240);
						newByteBuffer.clear();
						
						//获取返回数据
						String response = "result ok";
						newByteBuffer.put(response.getBytes("utf-8"));
						newByteBuffer.put((byte)'9');
						
						//注册写事件
						socketChannel.register(selector, SelectionKey.OP_WRITE, newByteBuffer);
						//----------------------------------------------------------------//
						
						
					} catch (IOException ioe) {
						ioe.printStackTrace();
						keyIterator.remove();
						socketChannel.close();
						continue;
					}
					
				} else if (key.isWritable()) {
					//通过key反向获取到对应的channel
					SocketChannel socketChannel = (SocketChannel)key.channel();
					//获取到该channel关联的buffer
					ByteBuffer byteBuffer = (ByteBuffer)key.attachment();
					
					//读写切换
					byteBuffer.flip(); //只有buffer出数据需要切换
					
					//把buffer数据写入到channel
					try {
						socketChannel.write(byteBuffer);
					} catch (IOException ioe) {
						ioe.printStackTrace();
						keyIterator.remove();
						socketChannel.close();
						continue;
					}
					
					String response = new String(byteBuffer.array(), 0, byteBuffer.position(), "utf-8");
					System.out.println("to " + socketChannel.hashCode() + " 服务端:" + response);
					
					//如果长连接继续注册事件等待
					//注册读事件
					//socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(10240));
					//如果短连接则断开
					socketChannel.close();
					
				} else if (key.isConnectable()) {
					System.out.println("Connectable");
				} else if (key.isValid()) {
					System.out.println("Valid");
				}
				
				//手动从集合中移除当前的SelectionKey,防止重复操作
				keyIterator.remove();
				
			}
			
		}
		
	}
}

NIOClient.java

package netty.niostart;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

public class NIOClient {
	public static void main(String[] args) throws Exception {
		//得到一个网络通道
		SocketChannel socketChannel = SocketChannel.open();
		//设置非阻塞模式
		socketChannel.configureBlocking(false);
		//提供服务器端的ip和端口
		InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 6666);
		
		//连接服务器
		if (!socketChannel.connect(inetSocketAddress)) {
			while(!socketChannel.finishConnect()) {
				System.out.println("因为连接需要时间,客户端不会阻塞,可以做其他工作");
			}
		}
		
		//如果连接成功,就发送数据
		String str = "111";
		//客户端也要关联buffer
		ByteBuffer byteBuffer = ByteBuffer.wrap(str.getBytes("utf-8")); //根据字节数组产生buffer
		
		//Thread.sleep(30*1000); //模拟线程阻塞
		
		//发送数据
		//将buffer数据写入channel
		socketChannel.write(byteBuffer);
		
		//获取返回
		byteBuffer.clear();
		
		int numBytesRead;
		while ((numBytesRead = socketChannel.read(byteBuffer)) != -1) { //-1是读完
			if (numBytesRead == 0) { //0是读到0个
				if (byteBuffer.limit() == byteBuffer.position()) {
                    byteBuffer.clear();
                }
				continue;
			}
			
			//buffer是数组用这个
			//System.out.println("客户端收到:" + new String(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.arrayOffset()+byteBuffer.position(), "utf-8"));
			//单个buffer用这个
			System.out.println("客户端收到:" + new String(byteBuffer.array(), 0, byteBuffer.position(), "utf-8"));
		}
		
		System.out.println("断开连接...");
		socketChannel.close();
		
		//System.in.read();
		
	}
}

二、用telnet测试

1、 打开cmd;
2、 执行:chcp65001,将编码方式改为“utf-8”;
3、 telnet连接;
4、 按Ctrl+];
5、 执行sendxxx;
6、 但是telnet传输字符长度有限制?;

三、服务端设想

 

1、 前端请求建立socket连接;
2、 网络通讯为NIO模型;
3、 将请求消息和socketChannel对象传到一个队列;
4、 业务模块维护一个线程池从队列获取请求并处理,还是一请求一线程模式;
5、 业务线程处理完成后注册写事件,结束;

这样网络读写是异步的,通讯和业务处理也是异步的