NIO

海龙2017

浏览: 60619 次
性别:
来自: 上海

最近访客更多访客>>

wangjungen

huxiulei

siemens800

redsnower

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

网络通讯技术java NIO
http://ifeve.com/java-nio-all/

产生背景
NIO(Non-blocking IO)非阻塞的IO模型

传统客户端-服务端每次新建公路用完后公路没了
现在客户端-服务端建立一个高速公路，每次都在这个公路上，
端口资源只占用一个。

阻塞：面向数据而言
同步：IO事件而晏
--现有IO模型
BIO-同步 jdk1.4之前数据缓冲区（水龙头下水桶）

NIO-同步 jak1.4以后非阻塞IO linux的多路复用技术，轮询机制
1 nio增加缓冲池。
2 高速通道Channel
3 线程模型方面改进 accept() 线程创建不断排队
4 增加多路复用机制 selector (一个线程+selector)
一个线程一个任务->任务注册selector，源码一个线程去轮训selector。
得到selectorKey一些就绪通道集合（底层源码）
epoll取代了传统select实现，没有最大的句柄限制。

AIO-异步 jdk1.7以后把IO读写交给操作系统学习linux epoll模式
[b]

NIO服务类介绍：
ServerSocketChannel 配合我们高速通道 Channel
Selector
SelectorKey

手写一个服务通信框架(NIO)并发

package com.hailong.yu.dongnaoxuexi;

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.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * 通信服务类
 * 接受上线客户端
 * @param args
 */
public class NIOService {

	private ServerSocketChannel server;
	int port =8080;
	//注册器
	private Selector selector;
	//数据缓冲区  接收
	ByteBuffer recBuffer =  ByteBuffer.allocate(1024);
	ByteBuffer sendBuffer =  ByteBuffer.allocate(1024);
	//中间缓存消息
	private Map<SelectionKey, String> sesionMaps =new HashMap<SelectionKey, String>();
	
	public NIOService(int port) throws IOException {
		this.port = port;
		server = ServerSocketChannel.open();
		// 服务类绑定监听端口
		server.socket().bind(new InetSocketAddress(port));
		// server要设置为非阻塞，执行任务时（在线程调度任务时），对方没有准备好数据，
		// 我们做其他任务，准备好后selector(selector中间调度器通知)，key事件得到通知，再继续做此任务。
		server.configureBlocking(false);// 默认是阻塞
		selector = selector.open();
		// 服务类和注册管家联系起来,服务类接收
		server.register(selector, SelectionKey.OP_ACCEPT);
		
		System.out.println("NIO服务已经启动，监听端口是"+this.port);
	}
	
	public void listenter() throws IOException{
		while(true) {
			int eventCount = selector.select();
			if(eventCount == 0) {
				continue;
			}
			// 这里就等于线程轮训selector拿到事件类型标签
			Set<SelectionKey> selectionKeySet = selector.selectedKeys();
			
			final Iterator<SelectionKey> iterator = selectionKeySet.iterator();
			while(iterator.hasNext()) {
				// 处理进程
				process(iterator.next());
			}
		} 
	}
	
	private void process(SelectionKey selectionKey) {
		SocketChannel channel = null;
		try {
			if(selectionKey.isValid() && selectionKey.isAcceptable()) {
				// 通过判断key 拿到链接进来channel
				channel =server.accept();
				channel.configureBlocking(false);
				channel.register(selector, SelectionKey.OP_READ);
			} else if(selectionKey.isValid() && selectionKey.isReadable()) {
				//读取客服端请求
				recBuffer.clear();
				channel = (SocketChannel)selectionKey.channel();
				int len = channel.read(recBuffer);
				if(len>0) {
					String msg = new String(recBuffer.array(),0,len);
					//聊天记录 维护key 缓存信息结构
					sesionMaps.put(selectionKey, msg);
					System.out.println("获取客户端发送来的信息：" +Thread.currentThread().getId()+ ","+ msg);
					channel.register(selector, SelectionKey.OP_WRITE);
				}
			} else if(selectionKey.isValid() && selectionKey.isWritable()) {
				
				if(!sesionMaps.containsKey(selectionKey)) {
					return;
				}
				
				channel = (SocketChannel)selectionKey.channel();
				//响应客服端请求
				sendBuffer.clear();
				sendBuffer.put(new String(sesionMaps.get(selectionKey)+",您好，处理完成").getBytes());
				//切换操作位
				sendBuffer.flip();
				channel.write(sendBuffer);
				channel.register(selector, SelectionKey.OP_READ);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			selectionKey.cancel();
			try {
				channel.socket().close();
				channel.close();
				return;
			} catch (IOException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
		}
	}
	
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		try {
			new NIOService(8080).listenter();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}

package com.hailong.yu.dongnaoxuexi;

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.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;

public class NIOClient {

	SocketChannel client;
	
	InetSocketAddress serverAddress = new InetSocketAddress("locallhost", 8080);
	
	Selector selector;
	
	//数据缓冲区  接收
	ByteBuffer recBuffer =  ByteBuffer.allocate(1024);
	ByteBuffer sendBuffer =  ByteBuffer.allocate(1024);
	
	public NIOClient() throws IOException {
		
		client = SocketChannel.open();
		client.configureBlocking(false);
		client.connect(serverAddress);
		selector = Selector.open();
		
		client.register(selector, SelectionKey.OP_CONNECT);
	}
	
	public void session() throws IOException {
		
		if(client.isConnectionPending()) {
			client.finishConnect();
			client.register(selector, SelectionKey.OP_WRITE);
			System.out.println("已经连接到服务器，可以在控制台登记了");
		}
		
		Scanner scan = new Scanner(System.in);
		while(scan.hasNextLine()) {
			//键盘输入内容
			String name =  scan.nextLine();
			if("".equals(name)) {
				continue;
			}
			if("finish".equals(name)) {
				System.exit(0);
			}
			process(name);
		}
	}
	
	private void process(String name) throws IOException {
		boolean waitHelp = true;
		
		Iterator<SelectionKey> iterator = null;
		Set<SelectionKey> keys = null;
		while(waitHelp) {
			int readys = selector.select();
			//如果没有客人，继续轮训
			if(readys ==0){
				continue;
			}
			keys = selector.selectedKeys();
			iterator = keys.iterator();
			try {
				while(iterator.hasNext()) {
					SelectionKey key = iterator.next();
					
					if(key.isValid() && key.isWritable()) {
						sendBuffer.clear();
						sendBuffer.put(name.getBytes());
						sendBuffer.flip();
						
						client.write(sendBuffer);
						client.register(selector, SelectionKey.OP_READ);
					} else if(key.isValid() && key.isReadable()) {
						recBuffer.clear();
						int len = client.read(recBuffer);
						if(len>0) {
							recBuffer.flip();
							System.out.println("服务器反馈消息"+new String(recBuffer.array(),0,len));
							client.register(selector, SelectionKey.OP_WRITE);
							waitHelp = false;
						}
					}
					iterator.remove();
				}
			} catch(IOException e) {

				((SelectionKey) keys).cancel();
				client.socket().close();
				client.close();
				return;

			}
		}
	}
	
	public static void main(String[] args) {
		// TODO Auto-generated method stub

	}

}

--NIO原理解读[/b]
1 多路复用(channel通道)
2 轮询机制（seletor选择器）注册行为给客户一个排队号（selectionKey），
处理完后，回收牌号
3 selectionKey
相当于时间标签
4 buffer数据缓冲区
5 NIO常用api
channel：
服务端：serverSocketChannel ,BIO:serverSocket
客户端：socketChannel ,BIO:socket
open();//建立公司修起来

channel.regist(selector, 注册的标识)

5.1.选择器 seclector
选择器有open方法 open(); 开始营业

selectKeys();//获取当前已经注册的所有牌号信息

5.2.缓冲区 Buffer
put() //往缓冲区里面写数据
get() //读数据
flip() //切换读写模式
clear() //清空缓冲区

5.3.牌号 seletionKey
isAccptable() //是否可以接受客户链接
isConctionable() //是否已经链接
isReadable() //缓冲区是否可读
isWriteabel() //是否可写

--java AIO简介
服务端： AsynchronousServerSocketChannel
客户端： AsynchronousServerChannel

CompletionMandler：通知程序，IO操作是否成功，还是失败。

--具体代码

分享到：