java nio 基础之Select 用法

虾米姐 阅读:562 2021-03-31 21:15:25 评论:0

本文参考至Java Nio 基础教程


Select (选择器)是Java Nio中能够检测一个到多个通道,并能够知道是否为诸如读写事件做好准备的组件。

这样,一个单独的线程就可以管理多个channel,从而管理多个网络连接。



为什么需要使用Select ?

仅用单个线程来处理多个Channels的好处是,只需要更少的线程来处理通道。事实上,可以只用一个线程处理所有的通道。对于操作系统来说,线程之间上下文切换的开销很大,而且每个线程都要占用系统的一些资源(如内存)。因此,使用的线程越少越好。

但是,需要记住,现代的操作系统和CPU在多任务方面表现的越来越好,所以多线程的开销随着时间的推移,变得越来越小了。实际上,如果一个CPU有多个内核,不使用多任务可能是在浪费CPU能力。不管怎么说,关于那种设计的讨论应该放在另一篇不同的文章中。在这里,只要知道使用Selector能够处理多个通道就足够了。


Selector 的创建

通过调用Selector.open()方法,创建Selector,代码如下:

Selector    selector = Selector.open();


向Selector 注册通道

为了将Channel 和Selector 配合使用,必须将Channel 注册到Selector 上,主要通过SelectableChannel.register()方法实现,代码如下:

serverSocketChannel.configBlocking(false);//启用通道非阻塞模式。

SelectionKey key = serverSocketChannel.register(selector,SelectionKey.OP_READ)


Selector 监听Channel 事件类型以及SelectionKey 常量表示。

Connect      :  通道触发了一个事件意思是该事件已经就绪。所以,某个channel成功连接到另一个服务器称为“连接就绪”     SelectionKey.OP_CONNECT

Accept        :  一个server socket channel准备好接收新进入的连接称为“接收就绪”。  SelectionKey.OP_ACCEPT

Read          :一个有数据可读的通道可以说是“读就绪”。     SelectionKey.OP_READ

Write           :等待写数据的通道可以说是“写就绪”。            SelectionKey.OP_WRITE


注意:如果你对不止一个事件感兴趣,可以使用“位或”操作符将常量联系起来

int interestset = SelectionKey.OP_ACCEPT | SelectionKey.OP_READ


SelectionKey 对象

当向selector 注册channel时,register()方法会返回一个SelectionKey 对象。这个对象包含以下几大属性。

1、interest 集合:intereset集合,是你选择感兴趣的事件集合。

2、ready集合:ready集合,是通道已经准备就绪的操作集合。

3、Channel:

4、Selector


interest集合:是你选择感兴趣的事件集合,通过SelectionKey.interesetOps()方法获取。代码如下:

1、int interestSet = selectionKey.interestOps();

2、boolean isInterestedInAccept  = (interestSet & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT;

3、boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;

4、boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;

5、boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;


ready集合:通道已经准备就绪的操作的集合。通过SelectionKey.readyOps()方法获取。代码如下:

int readyset = SelectionKey.readyOps();

可以用像检测interest集合那样的方法,来检测channel中什么事件或操作已经就绪。同时,也可以使用以下四个方法,它们都会返回一个布尔类型:

1、selectionKey.isAcceptable();

2、selectionKey.isConnectable();

3、selectionKey.isReadable();

4、selectionKey.isWritable();


Channel +Selector

从SelectionKey访问Channel和Selector很简单。代码如下:

Channel channel = SelectionKey.channel();

Selector selector = SelectionKey.selector();


通过Selector选择通道

一旦向Selector注册了一或多个通道,就可以调用几个重载的select()方法。这些方法返回你所感兴趣的事件(如连接、接受、读或写)已经准备就绪的那些通道。换句话说,如果你对“读就绪”的通道感兴趣,select()方法会返回读事件已经就绪的那些通道。

int select()   : 阻塞到至少有一个通道在你注册的事件上就绪了
int select(long timeout)  :  和select()一样,除了最长会阻塞timeout毫秒(参数)。
int selectNow() : 不会阻塞,不管什么通道就绪都立刻返回(译者注:此方法执行非阻塞的选择操作。如果自从前一次选择操作后,没有通道变成可选择的,则此方法直接返回零。)

注意重点
select()方法返回的int值表示有多少通道已经就绪。亦即,自上次调用select()方法后有多少通道变成就绪状态。如果调用select()方法,因为有一个通道变成就绪状态,返回了1,若再次调用select()方法,如果另一个通道就绪了,它会再次返回1。如果对第一个就绪的channel没有做任何操作,现在就有两个就绪的通道,但在每次select()方法调用之间,只有一个通道就绪了。


SelectedKeys()

一旦调用了select()方法,并且返回值表明有一个或更多个通道就绪了,然后可以通过调用selector的selectedKeys()方法,访问“已选择键集(selected key set)”中的就绪通道。如下所示: Set   selectedKeys  =  selector. selectedKeys()


当像Selector注册Channel时,Channel.register()方法会返回一个SelectionKey 对象。这个对象代表了注册到该Selector的通道。可以通过SelectionKey的selectedKeySet()方法访问这些对象。


可以遍历这个已选择的键集合来访问就绪的通道。代码如下:

Set selectedKeys = selector.selectedKeys(); 
Iterator keyIterator = selectedKeys.iterator(); 
while(keyIterator.hasNext()) { 
    SelectionKey key = keyIterator.next(); 
    if(key.isAcceptable()) { 
        // a connection was accepted by a ServerSocketChannel. 
    } else if (key.isConnectable()) { 
        // a connection was established with a remote server. 
    } else if (key.isReadable()) { 
        // a channel is ready for reading 
    } else if (key.isWritable()) { 
        // a channel is ready for writing 
    } 
    keyIterator.remove(); 
} 

这个循环遍历已选择键集中的每个键,并检测各个键所对应的通道的就绪事件。

注意每次迭代末尾的keyIterator.remove()调用。Selector不会自己从已选择键集中移除SelectionKey实例。必须在处理完通道时自己移除。下次该通道变成就绪时,Selector会再次将其放入已选择键集中。

SelectionKey.channel()方法返回的通道需要转型成你要处理的类型,如ServerSocketChannel或SocketChannel等。


示例demo:实现服务端与客户端消息传递。

package com.nio.two; 
 
import java.io.IOException; 
import java.net.InetSocketAddress; 
import java.nio.ByteBuffer; 
import java.nio.CharBuffer; 
import java.nio.channels.ClosedChannelException; 
import java.nio.channels.SelectionKey; 
import java.nio.channels.Selector; 
import java.nio.channels.ServerSocketChannel; 
import java.nio.channels.SocketChannel; 
import java.nio.charset.Charset; 
import java.nio.charset.CharsetDecoder; 
import java.nio.charset.CharsetEncoder; 
import java.util.Iterator; 
 
public class TCPServer { 
	private ByteBuffer readBuffer; 
    private Selector selector; 
     
    //编码器初始化 
    private static CharsetEncoder encoder = Charset.forName("utf-8").newEncoder(); 
    private static CharsetDecoder decoder = Charset.forName("utf-8").newDecoder(); 
    //初始化方法 
    private void init(){ 
        readBuffer = ByteBuffer.allocate(1024); 
        ServerSocketChannel servSocketChannel; 
          
        try { 
            servSocketChannel = ServerSocketChannel.open(); 
            servSocketChannel.configureBlocking(false); 
            //绑定端口 
            servSocketChannel.socket().bind(new InetSocketAddress(8383)); 
             //Select 创建 
            selector = Selector.open(); 
            //向Selector 注册通道(用户监控单一事件) 
            servSocketChannel.register(selector, SelectionKey.OP_ACCEPT); 
            //向Selector 注册通道(用户监控多个事件) 
//            int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE | SelectionKey.OP_ACCEPT | SelectionKey.OP_CONNECT; 
//            servSocketChannel.register(selector, interestSet); 
//            servSocketChannel.register(selector, SelectionKey.OP_ACCEPT); 
        } catch (IOException e) { 
            e.printStackTrace(); 
        }       
    } 
    //监听器 
    private void listen() { 
        while(true){ 
            try{ 
                selector.select();              
                Iterator ite = selector.selectedKeys().iterator(); 
                  
                while(ite.hasNext()){ 
                    SelectionKey key = (SelectionKey) ite.next();                   
                    ite.remove();//确保不重复处理 
                      
                    handleKey(key); 
                } 
            } 
            catch(Throwable t){ 
                t.printStackTrace(); 
            }                           
        }               
    } 
    //处理方法 
    private void handleKey(SelectionKey key) 
            throws IOException, ClosedChannelException { 
        SocketChannel channel = null; 
          
        try{ 
            if(key.isAcceptable()){ 
                ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); 
                channel = serverChannel.accept();//接受连接请求 
                channel.configureBlocking(false); 
                channel.register(selector, SelectionKey.OP_READ); 
            } 
            else if(key.isReadable()){ 
                channel = (SocketChannel) key.channel(); 
                readBuffer.clear(); 
                /*当客户端channel关闭后,会不断收到read事件,但没有消息,即read方法返回-1 
                 * 所以这时服务器端也需要关闭channel,避免无限无效的处理*/               
                int count = channel.read(readBuffer); 
                  
                if(count > 0){ 
                    //一定需要调用flip函数,否则读取错误数据 
                    readBuffer.flip(); 
                    /*使用CharBuffer配合取出正确的数据 
                    String question = new String(readBuffer.array());   
                    可能会出错,因为前面readBuffer.clear();并未真正清理数据 
                    只是重置缓冲区的position, limit, mark, 
                    而readBuffer.array()会返回整个缓冲区的内容。 
                    decode方法只取readBuffer的position到limit数据。 
                    例如,上一次读取到缓冲区的是"where", clear后position为0,limit为 1024, 
                    再次读取“bye"到缓冲区后,position为3,limit不变, 
                    flip后position为0,limit为3,前三个字符被覆盖了,但"re"还存在缓冲区中, 
                    所以 new String(readBuffer.array()) 返回 "byere", 
                    而decode(readBuffer)返回"bye"。             
                    */ 
                    CharBuffer charBuffer = decoder.decode(readBuffer);  
                    String question = charBuffer.toString();  
                    System.out.println("receiver client message:"+question); 
                    String answer = "the message come from this server"; 
                    channel.write(encoder.encode(CharBuffer.wrap(answer))); 
                } 
                else{ 
                    //这里关闭channel,因为客户端已经关闭channel或者异常了 
                    channel.close();                
                }                       
            }else if(key.isWritable()){ 
            	  channel = (SocketChannel) key.channel(); 
            	            	  
            	  String answer = "the message2 come from this server2"; 
                  channel.write(encoder.encode(CharBuffer.wrap(answer))); 
            }else if(key.isConnectable()){ 
            	 ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); 
                 channel = serverChannel.accept();//接受连接请求 
                 channel.configureBlocking(false); 
                 channel.register(selector, SelectionKey.OP_WRITE); 
            } 
        } 
        catch(Throwable t){ 
            t.printStackTrace(); 
            if(channel != null){ 
                channel.close(); 
            } 
        }       
    } 
	public static void main(String[] args) { 
		// TODO Auto-generated method stub 
		TCPServer server = new TCPServer(); 
		server.init(); 
		server.listen(); 
	} 
 
} 

package com.nio.two; 
 
import java.io.IOException; 
import java.net.InetSocketAddress; 
import java.nio.ByteBuffer; 
import java.nio.CharBuffer; 
import java.nio.channels.SelectionKey; 
import java.nio.channels.Selector; 
import java.nio.channels.SocketChannel; 
import java.nio.charset.Charset; 
import java.nio.charset.CharsetDecoder; 
import java.nio.charset.CharsetEncoder; 
import java.util.Iterator; 
 
 
public class TCPClient { 
	  //编码器初始化 
    private static CharsetEncoder encoder = Charset.forName("utf-8").newEncoder(); 
    private static CharsetDecoder decoder = Charset.forName("utf-8").newDecoder(); 
	public static void main(String[] args) { 
		// TODO Auto-generated method stub 
		  SocketChannel channel = null; 
	        Selector selector = null; 
	        try { 
	            channel = SocketChannel.open(); 
	            channel.configureBlocking(false); 
	            //请求连接 
	            channel.connect(new InetSocketAddress("localhost", 8383)); 
	            selector = Selector.open(); 
	            channel.register(selector, SelectionKey.OP_CONNECT); 
	            boolean isOver = false; 
	              
	            while(! isOver){ 
	                selector.select(); 
	                Iterator ite = selector.selectedKeys().iterator(); 
	                while(ite.hasNext()){ 
	                    SelectionKey key = (SelectionKey) ite.next(); 
	                    ite.remove(); 
	                      
	                    if(key.isConnectable()){ 
	                        if(channel.isConnectionPending()){ 
	                            if(channel.finishConnect()){ 
	                                //只有当连接成功后才能注册OP_READ事件 
	                                key.interestOps(SelectionKey.OP_READ); 
	                                  
	                                channel.write(encoder.encode(CharBuffer.wrap("Hello,Server!I come from Client"))); 
	                                 
	                            } 
	                            else{ 
	                                key.cancel(); 
	                            } 
	                        }                                               
	                    } 
	                    else if(key.isReadable()){ 
	                        ByteBuffer byteBuffer = ByteBuffer.allocate(128);                        
	                        channel.read(byteBuffer); 
	                        byteBuffer.flip(); 
	                        CharBuffer charBuffer = decoder.decode(byteBuffer); 
	                        String answer = charBuffer.toString();  
	                        System.out.println(Thread.currentThread().getId() + "---" + answer); 
	                          
	                        String word = "I have receiver message come from remote server"; 
	                        if(word != null){ 
	                            channel.write(encoder.encode(CharBuffer.wrap(word))); 
	                        } 
	                        else{ 
	                            isOver = true; 
	                        } 
	                                             
	                    } 
	                } 
	            }                           
	        } catch (IOException e) { 
	            e.printStackTrace(); 
	        } 
	        finally{ 
	            if(channel != null){ 
	                try { 
	                    channel.close(); 
	                } catch (IOException e) {                       
	                    e.printStackTrace(); 
	                }                   
	            } 
	              
	            if(selector != null){ 
	                try { 
	                    selector.close(); 
	                } catch (IOException e) { 
	                    e.printStackTrace(); 
	                } 
	            } 
	        } 
	} 
 
} 


小项目实战:基于java nio  实现简单聊天功能。

package com.nio.chart; 
 
import java.io.IOException; 
import java.net.InetSocketAddress; 
import java.nio.ByteBuffer; 
import java.nio.channels.Channel; 
import java.nio.channels.SelectionKey; 
import java.nio.channels.Selector; 
import java.nio.channels.ServerSocketChannel; 
import java.nio.channels.SocketChannel; 
import java.nio.charset.Charset; 
import java.util.HashSet; 
import java.util.Iterator; 
import java.util.Set; 
 
public class ChatRoomServer { 
	private Selector selector = null; 
	static final int port = 9999; 
	private Charset charset = Charset.forName("UTF-8"); 
	// 用来记录在线人数,以及昵称 
	private static HashSet<String> users = new HashSet<String>(); 
 
	private static String USER_EXIST = "system message: user exist, please change a name"; 
	// 相当于自定义协议格式,与客户端协商好 
	private static String USER_CONTENT_SPILIT = "#@#"; 
 
	private static boolean flag = false; 
 
	public void init() throws IOException { 
		selector = Selector.open(); 
		ServerSocketChannel server = ServerSocketChannel.open(); 
		server.bind(new InetSocketAddress(port)); 
		// 非阻塞的方式 
		server.configureBlocking(false); 
		// 注册到选择器上,设置为监听状态 
		server.register(selector, SelectionKey.OP_ACCEPT); 
 
		System.out.println("Server is listening now..."); 
 
		while (true) { 
			int readyChannels = selector.select(); 
			if (readyChannels == 0) 
				continue; 
			Set selectedKeys = selector.selectedKeys(); // 可以通过这个方法,知道可用通道的集合 
			Iterator keyIterator = selectedKeys.iterator(); 
			while (keyIterator.hasNext()) { 
				SelectionKey sk = (SelectionKey) keyIterator.next(); 
				keyIterator.remove(); 
				dealWithSelectionKey(server, sk); 
			} 
		} 
	} 
 
	public void dealWithSelectionKey(ServerSocketChannel server, SelectionKey sk) throws IOException { 
		if (sk.isAcceptable()) { 
			SocketChannel sc = server.accept(); 
			// 非阻塞模式 
			sc.configureBlocking(false); 
			// 注册选择器,并设置为读取模式,收到一个连接请求,然后起一个SocketChannel,并注册到selector上,之后这个连接的数据,就由这个SocketChannel处理 
			sc.register(selector, SelectionKey.OP_READ); 
 
			// 将此对应的channel设置为准备接受其他客户端请求 
			sk.interestOps(SelectionKey.OP_ACCEPT); 
			System.out.println("Server is listening from client :" + sc.getRemoteAddress()); 
			sc.write(charset.encode("Please input your name.")); 
		} 
		// 处理来自客户端的数据读取请求 
		if (sk.isReadable()) { 
			// 返回该SelectionKey对应的 Channel,其中有数据需要读取 
			SocketChannel sc = (SocketChannel) sk.channel(); 
			ByteBuffer buff = ByteBuffer.allocate(1024); 
			StringBuilder content = new StringBuilder(); 
			try { 
				while (sc.read(buff) > 0) { 
					buff.flip(); 
					content.append(charset.decode(buff)); 
 
				} 
				System.out.println( 
						"Server is listening from client " + sc.getRemoteAddress() + " data rev is: " + content); 
				// 将此对应的channel设置为准备下一次接受数据 
				sk.interestOps(SelectionKey.OP_READ); 
			} catch (IOException io) { 
				sk.cancel(); 
				if (sk.channel() != null) { 
					sk.channel().close(); 
				} 
			} 
			if (content.length() > 0) { 
				String[] arrayContent = content.toString().split(USER_CONTENT_SPILIT); 
				// 注册用户 
				if (arrayContent != null && arrayContent.length == 1) { 
					String name = arrayContent[0]; 
					if (users.contains(name)) { 
						sc.write(charset.encode(USER_EXIST)); 
 
					} else { 
						users.add(name); 
						int num = OnlineNum(selector); 
						String message = "welcome " + name + " to chat room! Online numbers:" + num; 
						BroadCast(selector, null, message); 
					} 
				} 
				// 注册完了,发送消息 
				else if (arrayContent != null && arrayContent.length > 1) { 
					String name = arrayContent[0]; 
					String message = content.substring(name.length() + USER_CONTENT_SPILIT.length()); 
					message = name + " say " + message; 
					if (users.contains(name)) { 
						// 不回发给发送此内容的客户端 
						BroadCast(selector, sc, message); 
					} 
				} 
			} 
 
		} 
	} 
 
	// TODO 要是能检测下线,就不用这么统计了 
	public static int OnlineNum(Selector selector) { 
		int res = 0; 
		for (SelectionKey key : selector.keys()) { 
			Channel targetchannel = key.channel(); 
 
			if (targetchannel instanceof SocketChannel) { 
				res++; 
			} 
		} 
		return res; 
	} 
 
	public void BroadCast(Selector selector, SocketChannel except, String content) throws IOException { 
		// 广播数据到所有的SocketChannel中 
		for (SelectionKey key : selector.keys()) { 
			Channel targetchannel = key.channel(); 
			// 如果except不为空,不回发给发送此内容的客户端 
			if (targetchannel instanceof SocketChannel && targetchannel != except) { 
				SocketChannel dest = (SocketChannel) targetchannel; 
				dest.write(charset.encode(content)); 
			} 
		} 
	} 
 
	public static void main(String[] args) throws IOException { 
		new ChatRoomServer().init(); 
	} 
 
} 

package com.nio.chart; 
 
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.nio.charset.Charset; 
import java.util.Iterator; 
import java.util.Scanner; 
import java.util.Set; 
 
 
public class ChatRoomClient { 
 
    private Selector selector = null; 
    static final int port = 9999; 
    private Charset charset = Charset.forName("UTF-8"); 
    private SocketChannel sc = null; 
    private String name = ""; 
    private static String USER_EXIST = "system message: user exist, please change a name"; 
    private static String USER_CONTENT_SPILIT = "#@#"; 
     
    public void init() throws IOException 
    { 
        selector = Selector.open(); 
        //连接远程主机的IP和端口 
        sc = SocketChannel.open(new InetSocketAddress("127.0.0.1",port)); 
        sc.configureBlocking(false); 
        sc.register(selector, SelectionKey.OP_READ); 
        //开辟一个新线程来读取从服务器端的数据 
        new Thread(new ClientThread()).start(); 
        //在主线程中 从键盘读取数据输入到服务器端 
        Scanner scan = new Scanner(System.in); 
        while(scan.hasNextLine()) 
        { 
            String line = scan.nextLine(); 
            if("".equals(line)) continue; //不允许发空消息 
            if("".equals(name)) { 
                name = line; 
                line = name+USER_CONTENT_SPILIT; 
            } else { 
                line = name+USER_CONTENT_SPILIT+line; 
            } 
            sc.write(charset.encode(line));//sc既能写也能读,这边是写 
        } 
         
    } 
    private class ClientThread implements Runnable 
    { 
        public void run() 
        { 
            try 
            { 
                while(true) { 
                    int readyChannels = selector.select(); 
                    if(readyChannels == 0) continue;  
                    Set selectedKeys = selector.selectedKeys();  //可以通过这个方法,知道可用通道的集合 
                    Iterator keyIterator = selectedKeys.iterator(); 
                    while(keyIterator.hasNext()) { 
                         SelectionKey sk = (SelectionKey) keyIterator.next(); 
                         keyIterator.remove(); 
                         dealWithSelectionKey(sk); 
                    } 
                } 
            } 
            catch (IOException io) 
            {} 
        } 
 
        private void dealWithSelectionKey(SelectionKey sk) throws IOException { 
            if(sk.isReadable()) 
            { 
                //使用 NIO 读取 Channel中的数据,这个和全局变量sc是一样的,因为只注册了一个SocketChannel 
                //sc既能写也能读,这边是读 
                SocketChannel sc = (SocketChannel)sk.channel(); 
                 
                ByteBuffer buff = ByteBuffer.allocate(1024); 
                String content = ""; 
                while(sc.read(buff) > 0) 
                { 
                    buff.flip(); 
                    content += charset.decode(buff); 
                } 
                //若系统发送通知名字已经存在,则需要换个昵称 
                if(USER_EXIST.equals(content)) { 
                    name = ""; 
                } 
                System.out.println(content); 
                sk.interestOps(SelectionKey.OP_READ); 
            } 
        } 
    } 
     
     
     
    public static void main(String[] args) throws IOException 
    { 
        new ChatRoomClient().init(); 
    } 
} 









声明

1.本站遵循行业规范,任何转载的稿件都会明确标注作者和来源;2.本站的原创文章,请转载时务必注明文章作者和来源,不尊重原创的行为我们将追究责任;3.作者投稿可能会经我们编辑修改或补充。

关注我们

一个IT知识分享的公众号