NIO 简单封装

java12345678

浏览: 355854 次
性别:
来自: 沈阳

最近访客更多访客>>

Franciswmf

cpongo1

java1234567

飘逸的霸地刺客

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

JAVA-NIO

对NIO进行封装，使其方便使用。

UML类图：

DatagramChannel

打开一个指定端口或随机端口，并可以接收任何发到本端口的数据报，也可以发送数据报到任何一个远程的地址。一个 DatagramChannel 即可以做为Server端（前提是Client端知道你的端口）也可以做为Client端，即 N对N的关系。

ServerSocketChannel

Socket通信中的服务端，打开一个指定的端口，可以接收多个Socket连接，如A主机发送Socket连接请求，连接成功后，生成SocketChannel ,该通道只能接收或发送数据到A主机。 N个主机发送N个Socket请求，连接成功后，服务端能够与N个主机通信。即1对N的关系。

SocketChannel

Socket通信中的客户端，打开一个指定端口或随机端口，连接到服务端后，就可以与服务端通信了。一条SocketChannel通道与服务端是 1对1的关系。

分别对上面三个通道进行封状，方便操作。

数据的接收：

使用NIO的selector监控通道的可读取事件（需要有一条线程来执行，暂称为监控线程），当有数据发来，获取发生事件的通道，并调用通道的读取方法获取数据，对于获取的数据如何处理呢？定义了一个数据接收器NetReceiver，方法onReceive(ReceiveData receiveData) 用于接收数据，具体的数据处理逻辑由该子类实现。

那么如何调用 NetReceiver. onReceive方法？？如果在当前监控线程中调用NetReceiver. onReceive，业务逻辑执行时间长，会影响该通道或其它通道上数据的接收延迟。因此需要单独的线程来执行。那么引入线程池，在监控线程中调用线程池 execute方法：

execute(new Runnable(){

NetReceiver. onReceive（接收到的数据）

})，这样线程池就可以并行处理接收到的数据提高处理速度。

存在一个问题：

先来说接收到的数据：对于DatagramChannel而言，receive方法接收会接收到一个完整数据报或返回，null表示当前没有可用的数据。接收到的数据以数据报为单位，理论最大字节数可以是65535（含包头）

对于SocketChannel而言,read方法可能读取0个或N个字节流。

对于 DatagramChannel ，如果接收到的每一个数据报，用一条线程来执行，如果前后数据报间没有顺关系是可以，但如果有顺序关系，如一个视频，一起接收到10个数据报并扔入线程池运行，可能第一个数据报最后执行完，因为线程的调度顺序与执行时间不能确定。这就导至了顺序的不一至。

那么可否10条线程进行同步，按顺序号来获取同步锁执行呢？这导至了9条线程等待，浪费线程。

最好是只有一线程执行。

对于 SocketChannel 而言，每一次接收到的数据字节数不确定，如果整个字节流中含多个业务数据单位，那么就需要哪些部分是一个业务数据单位，就需要每次接收的数据累加，直至形成一个完整的业务数据单位，再交给一条线程池执行，但需要多个业务数据单位间没有依赖关系。

因此生成三种线程模式：

一个对端址的发来的数据用一条线程处理，顺序流

一个数据报用一条线程来处理，多个数据报多个线程，第一个数据报都是独立了，与其它数据报无依赖

一个业务数据单位一条线程来处理，需要业务层提供判断如果判断接收到的数据够一个数据单元了。

主要接口：

/**
 *               
 * NIO网络通道的封装               
 * @author yanlei
 *
 */
public interface NetChannelWrap extends Runnable{
	
	/**
	 * 1.打开所有需要使用的资源，如通道或选择器或绑定端口等等。
	 * 2.监控通道的接收数据
	 * 3.当接收到数据时，调用NetReceiver.onReceiver方法
	 * @throws Exception
	 */
	public void open() throws IOException;

	/**
	 * 关闭网络通道,释放所有资源
	 */
	public void close() ;
	
	
}

/**
 * 数据报的网络通道封装
 * 
 * 1.当数据报通道异常时
 *    a.选择器上注销数据报通道，即不接收通道上的数据
 *   b.获取所有未处理完成的数据
 *   c.调用NetReceiverForSocket.onChannelErrorClose(e, receiveDataQueue);通知业务层所有网络通道不可用
 *   d.关闭所有通道
 * 
 * 2.当主动关闭数据报通道
 * 
 *    a.选择器上注销数据报通道，即不接收通道上的数据
 *    b.等待所有读取数据处理完成
 *    c.调用NetReceiver.onChannelClose()通知业务层，所有通道关闭
 *    d.关闭所有相关资源
 *    
 * @author yanlei
 *
 */
public interface  DatagramChannelWrap extends NetChannelWrap{
	public boolean send(SocketAddress target , ByteBuffer ...buffs ) throws  IOException;
	
}

/**
 * 
 * serverSocket通道的封装，用于数据通信中的服务端
 * 
 * 
 * 有几种情况需要注意：
 * 
 * 1.某条socket通道被服务端关闭（达到流的末尾）
 *    a.不在选择器上监控读取事件，即不接收读取
 *    b.获取该通道上所有接收的但未处理完成的数据
 *    c.通知业务层，该通该异常关闭了,即调用NetReceiverForSocket.onSocketChannelCloseByRemote(SocketAddress address,Queue<ReceiveData> receiveDataQueue);
 *      并传未处理未处理完的数据
 *    d. 关闭通道 
 *    
 * 2.某条socket通道发生异常
 *    a.不在选择器上监控读取事件，即不接收读取
 *    b.获取该通道上所有接收的但未处理完成的数据
 *    c.通知业务层，该通该异常关闭了,即调用NetReceiverForSocket.onSocketChannelErrorClose (IOException e,SocketAddress address,Queue<ReceiveData> receiveDataQueue);
 *      并传未处理未处理完的数据
 *    d. 关闭通道 
 *    
 * 3.客户端主动关闭某条socket通道
 * 	   a.在选择器上注销该通道的监控读取事件，即不接收读取
 *     b.等待该通道上所有的已接收数据处理完成
 *     c.通知业务层，该通该关闭了,即调用NetReceiverForSocket.onSocketChannelClose(SocketAddress address)
 *     d.关闭通道
 *     
 * 4.客户 端主动关闭所有通道
 *    a.选择器上注销所有通道，即不接收任务通道上的数据
 *    b.等待每一条通道上读取数据处理完成
 *    c.调用NetReceiverForSocket.onChannelClose()通知业务层，所有通道关闭
 *    d.关闭所有相关资源
 *    
 * 5.活动通道数等于0 ，  每一条通道或异常，或主动关闭或被动关闭，导致所有通道都关闭
 *   a.调用NetReceiverForSocket.haveNoActiveChannel()通知业务层，没有活动通道
 *   
 * 6.当选择器异常时
 *   a.选择器上注销所有通道，即不接收任务通道上的数据
 *   b.获取所有通道上的未处理完成的数据
 *   c.调用NetReceiverForSocket.onChannelErrorClose(e, receiveDataQueue);通知业务层所有网络通道不可用
 *   d.关闭所有通道
 *   
 * 
 * @author yanlei
 *
 */
public interface ServerSocketChannelWrap extends NetChannelWrap{
	public boolean send(SocketAddress target , ByteBuffer ...buffs ) throws  IOException;
	public void  closeSocketChannel(SocketAddress target);
}

/**
 * 
 * socket通道的封装，用于客户端连接到服务端通信
 * 
 * 支持一条或几条socket通道连接到一个服务端
 * 
 * 有几种情况需要注意：
 * 
 * 1.某条socket通道被服务端关闭（达到流的末尾）
 *    a.不在选择器上监控读取事件，即不接收读取
 *    b.获取该通道上所有接收的但未处理完成的数据
 *    c.通知业务层，该通该异常关闭了,即调用NetReceiverForSocket.onSocketChannelCloseByRemote(SocketAddress address,Queue<ReceiveData> receiveDataQueue);
 *      并传未处理未处理完的数据
 *    d. 关闭通道 
 *    
 * 2.某条socket通道发生异常
 *    a.不在选择器上监控读取事件，即不接收读取
 *    b.获取该通道上所有接收的但未处理完成的数据
 *    c.通知业务层，该通该异常关闭了,即调用NetReceiverForSocket.onSocketChannelErrorClose (IOException e,SocketAddress address,Queue<ReceiveData> receiveDataQueue);
 *      并传未处理未处理完的数据
 *    d. 关闭通道 
 *    
 * 3.客户端主动关闭某条socket通道
 * 	   a.在选择器上注销该通道的监控读取事件，即不接收读取
 *     b.等待该通道上所有的已接收数据处理完成
 *     c.通知业务层，该通该关闭了,即调用NetReceiverForSocket.onSocketChannelClose(SocketAddress address)
 *     d.关闭通道
 * 4.客户 端主动关闭所有通道
 *    a.选择器上注销所有通道，即不接收任务通道上的数据
 *    b.等待每一条通道上读取数据处理完成
 *    c.调用NetReceiverForSocket.onChannelClose()通知业务层，所有通道关闭
 *    d.关闭所有相关资源
 *    
 * 5.活动通道数等于0 ，  每一条通道或异常，或主动关闭或被动关闭，导致所有通道都关闭
 *   a.调用NetReceiverForSocket.haveNoActiveChannel()通知业务层，没有活动通道
 *   
 * 6.当选择器异常时
 *   a.选择器上注销所有通道，即不接收任务通道上的数据
 *   b.获取所有通道上的未处理完成的数据
 *   c.调用NetReceiverForSocket.onChannelErrorClose(e, receiveDataQueue);通知业务层所有网络通道不可用
 *   d.关闭所有通道
 *   
 * 
 * @author yanlei
 *
 */
public interface SocketChannelWrap extends NetChannelWrap {
	public boolean send(ByteBuffer ...buffs ) throws  IOException;
	public boolean send(SocketAddress fromLocalAddress , ByteBuffer ...buffs ) throws  IOException;
	public void  closeSocketChannel(SocketAddress target);
}

/**
 * 网络数据接收器
 * 1.用于接收网络发来的数据
 * 2.网络通道或异常时被触发
 * @author yanlei
 *
 */
public interface  NetReceiver {
	
	/**
	 * 当接收到数据，触发本方法
	 * @param ReceiveData 数据对象
	 */
	public void onReceive(ReceiveData receiveData);

	
	/**
	 *  当本地端口的网络通道或选择器发生异常时，调用本方法，并自动关闭网络通道
	 * @param e 异常
	 * @param receiveDataList 已接收到，但未调用onReceive方法处理的数据
	 */
	public void onChannelErrorClose(IOException e,Queue<ReceiveData> receiveDataQueue);
	
	/**
	 * 当本地端口 的网络通道被关闭时被调用
	 */
	public void onChannelClose();
	/**
	 * 获取接收数据线程的模式
	 * 一个对端地址使用一条线程：用于处理长的顺序流
	 * 一个数据包使用一条线程：数据包之间无关，用于数据报
	 * 一个个完整的数据段使用一条线程：用于socket累加接收数据，直至数据完整 ，才启用一条线程数据。
	 * @return
	 */
	public NetReceiveTaskMode getTaskMode();
	
	/**
	 * 当通道异常时，已接收到的但未处理的数据是否继续处理
	 * @return
	 */
	public boolean isDealReceiveOnChannelError();
}

/**
 *  socket网络连接的数据接收器
 * 
 * 1.用于接收网络发来的数据
 * 2.网络通道或异常时被触发        
 * 
 * 
 * @author yanlei
 *
 */
public interface NetReceiverForSocket extends NetReceiver{
	/**
	 *  当socket连接成功时触发
	 *  服务端 ：socketAddress指远程地址连接到本服务端
	 *  客服端：socketAddress指本地端口连接到远程服务端
	 * @param socketAddress
	 */
	public void onSocketConnect(SocketAddress socketAddress);
	/**
	 * 当某条socket通道被对端关闭时(流达到末尾)触发
	 * @param address 本端是服务端：address 指远程端口地址 ，本端是客户端：address指本地端口地址
	 * @param receiveDataQueue 已接收但未处理完成的接收数据列表
	 */
	public void onSocketChannelCloseByRemote(SocketAddress address,Queue<ReceiveData> receiveDataQueue);
	/**
	 * 当某条socket通道被调用closeSocketChannel方法被关闭时触发
	 * @param address
	 */
	public void onSocketChannelClose(SocketAddress address);
	/**
	 * 当某条socket通道发生异常时触发
	 * @param e 异常
	 * @param address 本端是服务端：address 指远程端口地址 ，本端是客户端：address指本地端口地址
	 * @param receiveDataQueue 已接收但未处理完成的接收数据列表
	 */
	public void onSocketChannelErrorClose (IOException e,SocketAddress address,Queue<ReceiveData> receiveDataQueue);
	/**
	 * 
	 * addUpBuffer中存的是累加的数据，newReceiveBuffer是新接收到的数据，将newReceiveBuffer的数据
	 * 累加到addUpBuffer中，直到数据完整，则返回完整的数据，addUpBuffer等于剩余的数据，并等待下一次的累加
	 * @param addUpBuffer 累加的数据
	 * @param newReceiveBuffer 新接收到的数据
	 * @return
	 */
	public List<ByteBuffer> addUpIntegrityBuffer (ByteBuffer addUpBuffer , ByteBuffer newReceiveBuffer) ;
	/**
	 * 当没有活动的socket通道时被触发
	 */
	public void haveNoActiveChannel();
	/**
	 * 定义最大的完整数据的字节大小
	 * @return
	 */
	public int getMaxAddUpBufferSize();
	
}

/**
 * 由线程池执行的线程任务接口，内部调用NetChannel.onReceive 处理业务逻辑
 * 主要解决问题
 *   1.dealReceiveData 用多线程执行接收数据处理，提高整体处理速度
 *   2.removeAllReceiveData 移除所有接收到但未处理的数据，用于异常时不再处理，将不处理的数据交给NetReceive接口，让业务层决定该数据如何处理
 *   3.setFinishRunable 设定当需要处理的数据都处理完成时，调用的接口，一般用于SocketChannel在所有接收数据处理完成后再关闭
 * @author yanlei
 *
 */
public interface  NetReceiveTask {

	public  void dealReceiveData(ReceiveData receiveData);
	public  Queue<ReceiveData> removeAllReceiveData();
	public  void setFinishRunable(Runnable call);
	
	
}

/**
 * 
 *     每一个数据报的对端地址或每一条socket连接，需要使用一条线程来顺序处理接收到的字节数据。
 *     解决顺序流问题，如果用多条线程执行，线程执行的先后顺序不通确定，
 *     可能后进来的数据优先处理完了，可能会影响数据结构。 
 * @author yanlei
 *
 */
public class NetReceiveTaskOneThread implements NetReceiveTask,Runnable{
	 ExecutorService executorService = null;
	
	 NetReceiver netReceiver = null;
	
	
		SocketAddress socketAddress = null;
    	Queue<ReceiveData> receiveDataQueue = new ConcurrentLinkedQueue<ReceiveData>();
    	
    	NetReceiveTaskOneThread(SocketAddress socketAddress,NetReceiver netReceiver,ExecutorService executorService){
    		this.socketAddress = socketAddress;
    		 this.executorService = executorService;
    		 this.netReceiver = netReceiver;
    	}
    	volatile boolean  stop = true;
    	public synchronized void dealReceiveData(ReceiveData receiveData){
    		this.receiveDataQueue.add(receiveData);
    		if(stop){
    			 this.executorService.execute(this);
    			stop = false;
    		}
    	}
    	public Queue<ReceiveData> removeAllReceiveData(){
    		Queue<ReceiveData> tempQueue = receiveDataQueue;
    		 receiveDataQueue = new ConcurrentLinkedQueue<ReceiveData>();
    		 return tempQueue;
    	}
    	Runnable finishCall = null;
    	public void setFinishRunable(Runnable call){
    		this.finishCall = call;
    	}
    	public synchronized void onFinish(){
    		if(this.receiveDataQueue.isEmpty()){
    			stop = true;
    			if(this.finishCall !=null){
    				try{
    					this.finishCall.run();
    				}catch(Exception e){
    					
    				}
    			}
    		}else{
    			this.executorService.execute(this);
    			stop = false;
    		}
    		
    	}
		@Override
		public void run() {
			// TODO Auto-generated method stub
			ReceiveData receiveData = receiveDataQueue.poll();
			
			while(receiveData != null){
				this.netReceiver.onReceive(receiveData);
				receiveData = receiveDataQueue.poll();
			}
			onFinish();
		}
}

/**
 * 
 * 每一次接收到的数据，都使用一条线程来处理，主用于无顺序的udp数据报，即一个
 *       数据包中包含了完整的数据结构，并前后数据报之间没有关联。
 * @author yanlei
 *
 */
public class NetReceiveTaskMultiThread  implements NetReceiveTask,Runnable{
	 ExecutorService executorService = null;
	
	 NetReceiver netReceiver = null;
	
	
		SocketAddress socketAddress = null;
    	Queue<ReceiveData> receiveDataQueue = new ConcurrentLinkedQueue<ReceiveData>();
    	NetReceiveTaskMultiThread(SocketAddress socketAddress,NetReceiver netReceiver,ExecutorService executorService){
    		this.socketAddress = socketAddress;
    		 this.executorService = executorService;
    		 this.netReceiver = netReceiver;
    	}
    	AtomicInteger num = new AtomicInteger(0);
    	public  void dealReceiveData(ReceiveData receiveData){
    		this.receiveDataQueue.add(receiveData);
    		num.incrementAndGet();
    		this.executorService.execute(this);
    			
    	}
    	public Queue<ReceiveData> removeAllReceiveData(){
    		Queue<ReceiveData> tempQueue = receiveDataQueue;
    		 receiveDataQueue = new ConcurrentLinkedQueue<ReceiveData>();
    		 return tempQueue;
    	}
    	Runnable finishCall = null;
    	public void setFinishRunable(Runnable call){
    		this.finishCall = call;
    	}
    	public  void onFinish(){
    		
    			if(this.finishCall !=null){
    				try{
    					this.finishCall.run();
    				}catch(Exception e){
    					
    				}
    			}
    		
    		
    	}
		@Override
		public void run() {
			// TODO Auto-generated method stub
			ReceiveData receiveData = receiveDataQueue.poll();
			
			if(receiveData != null){
				this.netReceiver.onReceive(receiveData);
			}
			if(num.decrementAndGet()==0){
				onFinish();
			}
		}
}

/**
 * 
 * 每一次接收到完整数据，使用一条线程来执行。
 *       从网络中接收到的数据，对于业务来讲，当前已接收到的数据可能是不完整的，需要再次接收数据，累加成完整的
 *       数据结构，才使用一条线程来执行。     
 *       
 *       完整数据的判断方式：NetReceiverForSocket.addUpIntegrityBuffer (ByteBuffer addUpBuffer , ByteBuffer newReceiveBuffer) ;
 * @author yanlei
 *
 */
public class NetReceiveTaskForSocketMultiThread extends NetReceiveTaskMultiThread{
		ByteBuffer addUpBuffer  = null;
    	NetReceiveTaskForSocketMultiThread(SocketAddress socketAddress,NetReceiverForSocket netReceiver,ExecutorService executorService){
    		super(socketAddress,netReceiver,executorService);
    		this.addUpBuffer = ByteBuffer.allocate(((NetReceiverForSocket)this.netReceiver).getMaxAddUpBufferSize());

    	}
    	
    	public  void dealReceiveData(ReceiveData receiveData){
    		List<ByteBuffer > integrityBufferList = ((NetReceiverForSocket)this.netReceiver).addUpIntegrityBuffer(this.addUpBuffer, receiveData.getByteBuffer());
    		if(integrityBufferList != null && integrityBufferList.size()>0){
    			for(ByteBuffer byteBuffer:integrityBufferList){
    				super.dealReceiveData(new ReceiveData(receiveData.getAddress(),byteBuffer,receiveData.getNetChannelWrap()));
    			}
    		}
    			
    	}
    
}

/**
 * 扩展线程池，添加shutdown方法，用于在线程池中所有任务执行完成时，调用Closeable接口，
 * Closeable接口实现类一般来关闭网络通道。即线程池执行完成后，达到关闭网络通道的目的
 * 
 * @author yanlei
 *
 */
public interface ReceiveThreadPool extends ExecutorService{
	/**
	 * 用于在线程池结束时执行指定的接口 
	 * @param closeable
	 */
	public void shutdown(Closeable closeable);
}