DatagramChannel

 

UDP 是User Datagram Protocol的简称，UDP进行通信，客户端与服务端不进行连接，只是使用数据报进行通信。

一个程序打开一个UDP端口，可以给任何远程主机的UDP端口发送数据报，也可以接收任何发送到本端口的数据报（前提是别人知道你的UDP端口）。 

 

数据报最大可以是65536字节。

 

DatagramChannel  可以打开并监听一个UDP端口，接收发送到该端口的任何数据报（可能来源于任何一台主机），也可以发送数据报给任何一台主机。（一个DatagramChannel 对应一个UDP端口）

 

 

API:

 

static DatagramChannel	open()           打开数据报通道。
abstract  SocketAddress	receive(ByteBuffer dst)           通过此通道接收数据报。
abstract  int	send(ByteBuffer src, SocketAddress target)           通过此通道发送数据报。
abstract  DatagramSocket	socket()           获取与此通道关联的数据报套接字。
int	validOps()           返回一个操作集，标识此通道所支持的操作，该值为5，即 SelectionKey.OP_READ 和OP_WRITE。Selector可监控数据报通的的可读与可写。                       事件类型由一个int值表示：                        第0位表示可读，第2位表示可写，第3位表示有连接请求，第4位表示可接收连接请求

 

 

 实例的产生：需要调用静态的open方法，获取DatagramChannel实例。

 

阻塞状态： DatagramChannel 打开（open）之后，默认为阻塞状态。

                  当调用read方法时，如果没有数据报发来，则一至阻塞直至接收到数据报才返回。

                  当调用write方法时，由于底层缓冲区没有足够空间容纳要发送的数据，则阻塞，直至缓冲区够大才发送。

非阻塞状态： DatagramChannel 打开（open）之后，调用configureBlocking（false）方法，将数据报通道 配置成为非阻塞模式。 

                      非阻塞模式需要与Selector(选择器)一起使用，将DatagramChannel（数据报通道）在Selector注册了感兴趣的事件（read/write），调用Selector.select()方法时，该方法会阻塞直至发生了感兴趣的事件，如接收到数据报（read）或可以发送数据报(write，底层缓冲区可用时)，select()方法结果并返回SelectionKey,通过SelectionKey的方法 isReadable()或 isWritable()来判断是哪一种操作准备好了，SelectionKey.channel()返回是哪条数据报通道准备好了（ Selector 上可以注册多个可选择通道如多个  DatagramChannel  或SocketChannel），再调用 该DatagramChannel的接收或发送方法。

 

                      非阻塞模式下，需要一条线程在无限循环中不断的调用 Selector.select() 以监控数据报通道的可读或可写。在selector只监控一条  DatagramChannel  情况下，与阻塞模式类似：需要一条线程调用receive方法，该方法阻塞直至有数据报发来，调用send方法时，可能由于底层缓冲区不够而阻塞直至缓冲区可用。

 

随机端口：默认为随机端口

非随机端口：使用  DatagramChannel.socket().bind(new InetSocketAddress(端口号))绑定到指定端口

 

 

 与任意地址通信：

SocketAddress receive(ByteBuffer dst)：该方法接收任意地址发来的数据报

      该方法在非阻塞模式下会被阻塞直至接收到数据报，在非阻塞模式下得到可读事件时再调用不会阻塞。 

      

       如果数据报中的字节数大于给定缓冲区中的剩余空间，则丢弃余下的数据报，数据报最大为65536字节，因此针对实际情况设置一个合适的缓冲区大小以防止数据丢弃。 可以创建一个足够大的 ByteBuffer 接收数据报，每次接收都使用这个ByteBuffer（只有一个）,接收完成后，按实际数据报大小再生成一个ByteBuffer,这个ByteBuffer会小得多，由业务逻辑代码对这个ByteBuffer进行处理，节省了内存又防止多线程同时对一个ByteBuffer处理引用线程安全问题。

       接收数据固定要使用一条线程，那么对接收到的数据报处理是在接收线程中或使用另外的一条或几条线程，取决于接收数据报的频率与业务逻辑处理时间的长短。最好接收一条线程，业务处理一条线程，数据少时只是多占用了一条线程，数据报频繁或业务逻辑处理时间长时，也可以方便扩充业务逻辑线程个数，对程序的改动少。

       返回值： SocketAddress ，指定数据报的来源，可以根所这个  SocketAddress  给对端发送数据报。

 

 

 

int send(ByteBuffer src, SocketAddress target) ：该方法发送数据报到指定的目标地址。

    将 ByteBuffer 中的可用数据（position至limit）打成单个数据报发给给目标地址。

 

    该方法在非阻塞模式下可能会被阻塞（可能是底层数据缓冲区不够大时）直至能把可用数据全部发送出去（一个数据报），返回全部数据的字节数。

    该方法在非阻塞模式下，可能因为底层数据缓冲区不够大，不能够容纳  ByteBuffer  中的可用数据（不能生成一个完整数据报）而返回0，即不发送任何字节。 策略是要么全部发送要么不发送，因为发送出的数据要打成单个数据报发送给目标地址。这与socket通道不同：socket通道是连接的，在通道上能够发送几个字节就发送几个字节，并返回发送字节个数，因此socket通道的发送方法需要在无限循环中，直至所有的字节发送完成才退出循环，即完成发送。

     一般非阻塞模式下，在Selector(选择器)中只注册读取事件，当调用send方法（写操作）返回0时，表明底层数据缓冲区不够大，那么什么时候缓冲区够大呢？不知道，那么不断的发送呢：

while((sendNum = this.datagramChannel.send(src, target))==0){};

这段代码在缓冲区足够大之前，一直空跑，占用cpu。

解决方式一：

    当 send方法返回0时，在Selector(选择器)上注册可写事件，当发生可写事件时，再调用send方法发送。

    这种方式可能会造成同一通道上的send的数据先后顺序改变：

        当选择器发现可以写入前，可能另一线程调用同一通道的send方法，这时刚好可以写入，之后选择器才发现可写入事件并写入数据，这就造成了，后调用send方法的数据比先调用方法的数据优先写入了。对于数据流顺序不影响业务时，可以这么做。

解决方式二：

       当 send方法返回0时，打开一个新的选择器，并注册监控该通道上的写入事件，调用该选择器上的select()方法，这时阻塞，直到可以写入了，再send.

 

 

//返回值 sendNum==length ,说明数据全被发送出去了，返回值sendNum=0，说明底层输出缓冲区中没有足够的空间供数据报，则未发送任何字节数据
			//与socketChannel 不同，socketChannel在底层输出缓冲区中没有足够的空间时，能发送几个字节就发几个字节。返回值为发送字节个数。
			if(sendNum ==0){
				 SendData sendData = new SendData();
				 sendData.setByteBuffer(allByteBuffer);
				 sendData.setTarget(target);
				 this.sendDataQueue.add(sendData);//加入待发队列
				 SelectionKey key = datagramChannel.keyFor(this.selector);//获取注册的key
				 key.interestOps(SelectionKey.OP_READ |SelectionKey.OP_WRITE);//加入监控可写事件
				 this.selector.wakeup();//唤醒选择器，使其在下一周期select()时监控可写事件，当前正执行的select()方法不会监控新加入的可写入事件
			}

 

 

 

while(true){
				
					selector.select();//阻塞直至收到数据包,返回值可能为零，但有事件发生，因此不以返回值判断事件数
					if(this.stop){//退出标志
						break;
					}
					
					Set<SelectionKey> keys = selector.selectedKeys();//获取发生读取事件的注册键 
					Iterator<SelectionKey> iterator = keys.iterator();
					while(iterator.hasNext()){//遍历
						 SelectionKey key=iterator.next();
						 iterator.remove();//需要手工移除注册键，否则下次selectedKeys里仍然包括它（虽然该selectionKey对应的通道上没有事件）
						// DatagramChannel dc = (DatagramChannel)key.channel();//获取接收数据通道==datagramChannel
						 if(key.isWritable()){//通道可以写入了
							 if(!sendDataQueue.isEmpty()){//重发队列不为空
								 SendData sendData = null;
								 while((sendData = sendDataQueue.peek()) != null){
									 if(!this.send(sendData.getTarget(), sendData.getByteBuffer())){//重发失败
										 break;
									 }else{
										 sendDataQueue.poll();//重发成功
									 }
								 }
								
							 }else{//重发队列空了,不再监控可写入事件，只监控接收数据事件
								 key.interestOps(SelectionKey.OP_READ);
							 }
							
						 }else if(key.isReadable()){//接收数据事件，有数据被接收了
							 byteBuffer.clear();
							 final SocketAddress from = this.receive(byteBuffer);//接收数据包,返回数据来源
							//处理业务逻辑
						}	 
  }						

 

 

  与固定地址通信：

 首先调用connect 方法连接指定远程地址，之后就可以调用不带有下面的read 或write方法与之通信。

 在调用disconnect ()断开连接以前，不能与其它地址通信。

read方法类似于 receive方法。

write方法类似于send方法。

 

abstract  DatagramChannel	connect(SocketAddress remote)           连接此通道的套接字。
abstract  DatagramChannel	disconnect()           断开此通道套接字的连接。
abstract  boolean	isConnected()           判断是否已连接此通道的套接字。

abstract  int	read(ByteBuffer dst)           从此通道读取数据报。
long	read(ByteBuffer[] dsts)           从此通道读取数据报。
abstract  long	read(ByteBuffer[] dsts, int offset, int length)           从此通道读取数据报。

abstract  int	write(ByteBuffer src)           将数据报写入此通道。
long	write(ByteBuffer[] srcs)           将数据报写入此通道。
abstract  long	write(ByteBuffer[] srcs, int offset, int length)           将数据报写入此通道。

 

    

 

 

 DatagramChannel 使用方法：

  datagramChannel  = DatagramChannel.open();//打开通道
	  datagramChannel.socket().bind(new InetSocketAddress(this.localPort));//绑定本地端口
	  datagramChannel.configureBlocking(false);//配置成非阻塞模式
	  selector = Selector.open();//打开选择器
	  datagramChannel.register(selector, SelectionKey.OP_READ);//注册监听可读取事件
	  while(true){
		  selector.select();//阻塞直至收到数据报,返回值可能为零，但有事件发生，因此不以返回值判断事件数
		  Set<SelectionKey> keys = selector.selectedKeys();//获取发生读取事件的注册键 
		  Iterator<SelectionKey> iterator = keys.iterator();
			while(iterator.hasNext()){//遍历
				 SelectionKey key=iterator.next();
				 iterator.remove();//需要手工移除注册键，否则下次selectedKeys里仍然包括它（虽然该selectionKey对应的通道上没有事件）
				// DatagramChannel dc = (DatagramChannel)key.channel();//获取接收数据通道==datagramChannel
				 if(key.isWritable()){//通道可以写入了
					 
				 }else if(key.isReadable()){//通道可以读取了
					 
				 }
			}	 
	  }