Java NIO 之 Selector
这篇博文将来记录下自己Java NIO中另一个重要的东西—-Selector。
关于Selector确实不太好理解，自己也看了很多博客，到目前为止，自己对Selector还是没有弄清楚，感觉模模糊糊的。最主要的可能是没有找到对应的应用场景，不知道在上面地方能够用到Selector。
关于Selector的相关介绍，并发编程网的这篇博文介绍的比较好：http://ifeve.com/selectors/。
本篇博文就不对Selector进行介绍了，本着学习的态度，自己也来尝试通过写Demo来理解下Selector，于是就有了这篇博文。
Selector例子
本应用设想的场景是这样的：
在服务器端：
1、首先将ServerSocketChannel注册在Selector中，并为此通道注册SelectionKey.OP_ACCEPT事件。
2、当有客户端连接进来之后，将SocketChannel注册到Selector中，并为此通道注册SelectionKey.OP_READ事件，便于在服务器端接收到客户端发送过来的数据。
在客户端：
1、首先将SocketChannel注册在Selector中，并为此通道注册SelectionKey.OP_CONNECT事件。
2、当客户端和服务器端连接成功之后，将其注册到Selector中，并为此通道注册SelectionKey.OP_READ事件，便于在客户端接收到服务器点发送过来的数据。
最后，实现客户端和服务器端之间的交互。
这个Demo具体的交互工程如下
客户端和服务器端建立连接后，客户端第一次发送 1，第二个发送2，第N次发送N；服务器端每接收一个消息就给客户端发送一个“收到消息”。
仿照这篇博文中的代码：http://blog.csdn.net/anders_zhuo/article/details/8535719
写了下代码，发现并没有实现如上的功能，有一定的问题，问题还没有找出来，明天继续来找。
实现代码如下：
package selector;
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 {
//通道选择器
private Selector selector;
public NIOServer(){
try {
selector = Selector.open();
} catch (IOException e) {
e.printStackTrace();
}
}
/*
* 函数功能：服务器端开始监听，看是否有客户端连接进来
* */
private void listen() throws IOException {
System.out.println("server running....");
while(true){
// 当注册事件到达时，方法返回，否则该方法会一直阻塞
selector.select();
// 获得selector中选中的相的迭代器，选中的相为注册的事件
Set<SelectionKey> set = selector.selectedKeys();
Iterator<SelectionKey> ite = set.iterator();
while(ite.hasNext()){
SelectionKey selectionKey = (SelectionKey) ite.next();
// 删除已选的key 以防重负处理
ite.remove();
if(selectionKey.isAcceptable()){//如果有客户端连接进来
//先拿到这个SelectionKey里面的ServerSocketChannel。
ServerSocketChannel serverSocketChannel = (ServerSocketChannel)selectionKey.channel();
// 获得和客户端连接的通道
SocketChannel socketChannel = serverSocketChannel.accept();
System.out.println("有客户端连接到服务器！！！");
socketChannel.configureBlocking(false);//将此通道设置为非阻塞
//服务器端向客户端发送数据
// ByteBuffer buf = ByteBuffer.allocate(40);
// buf.put("hello client!!!".getBytes());
// socketChannel.write(buf);
socketChannel.write(ByteBuffer.wrap(new String("hello client!").getBytes()));
//为了接收客户端发送过来的数据，需要将此通道绑定到选择器上，并为该通道注册读事件
socketChannel.register(selector, SelectionKey.OP_READ);
}
else if(selectionKey.isReadable()){//客户端发送数据过来了
//先拿到这个SelectionKey里面的SocketChannel。
SocketChannel socketChannel = (SocketChannel)selectionKey.channel();
//接收来自于客户端发送过来的数据
ByteBuffer buf = ByteBuffer.allocate(20);
int len = 0;
System.out.println("服务器端接收到的数据为：");
while((len=socketChannel.read(buf))!=-1){
// while(buf.hasRemaining()){
// System.out.println(buf.get()+" ");
// }
byte[] receData = buf.array();
String msg = new String(receData).trim();
System.out.println("接收来自客户端的数据为："+msg);
buf.clear();
}
//服务器端向客户端发送"确定信息"
buf.put("收到信息".getBytes());
while(buf.hasRemaining()){
socketChannel.write(buf);
}
//buf.clear();
}
}
}
}
/*
* 函数功能：初始化serverSocketChannel来监听指定的端口是否有新的TCP连接，
* 并将serverSocketChannel注册到selector中
* */
private void init(int port) {
try {
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
//serverSocketChannel监听指定端口
serverSocketChannel.socket().bind(new InetSocketAddress(port));
serverSocketChannel.configureBlocking(false);//设置为非阻塞模式
/*
* 将serverSocketChannel注册到selector中,并为该通道注册selectionKey.OP_ACCEPT事件
* 注册该事件后，当事件到达的时候，selector.select()会返回， 如果事件没有到达selector.select()会一直阻塞
* */
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
} catch (IOException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws IOException {
NIOServer server = new NIOServer();
server.init(9999);
server.listen();
}
}
客户端
package selector;
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.Set;
public class NIOClient {
private Selector selector;
public NIOClient() throws IOException {
this.selector =Selector.open();
}
private void init(String address,int port) throws IOException{
//客户端，首先有一个SocketChannel
SocketChannel socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);//将此通道设置为非阻塞模式
//连接
socketChannel.connect(new InetSocketAddress(address,port));
//将SocketChannel注册到selector中，并为该通道注册SelectionKey.OP_CONNECT
socketChannel.register(selector, SelectionKey.OP_CONNECT);
}
public static void main(String[] args) throws IOException {
NIOClient client = new NIOClient();
client.init("localhost",9999);
client.connect();
}
private void connect() throws IOException {
int data = 1;
while(true){
selector.select();//
Set<SelectionKey> set = selector.selectedKeys();
Iterator<SelectionKey> ite = set.iterator();
while(ite.hasNext()){
SelectionKey selectionKey = (SelectionKey) ite.next();
ite.remove(); //删除已选的key,以防重复处理
if(selectionKey.isConnectable()){//看是否有连接发生
SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
//如果正在连接，则完成连接
if(socketChannel.isConnectionPending()){
socketChannel.finishConnect();
}
socketChannel.configureBlocking(false);//设置为非阻塞模式
//给服务器端发送数据
System.out.println("客户端连接上了服务器端。。。。");
// ByteBuffer buf = ByteBuffer.allocate(128);
// buf.put("hello server.....".getBytes());
// socketChannel.write(buf);
socketChannel.write(ByteBuffer.wrap(new String("hello server!").getBytes()));
//为了接收来自服务器端的数据，将此通道注册到选择器中
socketChannel.register(selector, SelectionKey.OP_READ);
}
else if(selectionKey.isReadable()){
SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
//接收来自于服务器端发送过来的数据
ByteBuffer buf = ByteBuffer.allocate(10);
int len = 0;
System.out.println("客户端接收到的数据为：");
while((len=socketChannel.read(buf))!=-1){
// while(buf.hasRemaining()){
// System.out.println(buf.get()+" ");
// }
byte[] receData = buf.array();
String msg = new String(receData).trim();
System.out.println("接收来自服务器端的数据为："+msg);
buf.clear();
}
//发送数据
buf.put((data+"").getBytes());
while(buf.hasRemaining()){
socketChannel.write(buf);
}
data++;
}
}
}
}
}
Selector例子(修正)
在上面的例子中，就如昨天所说，还存在一点问题，客户端和服务器端还不能够正常的交互。2016年10月11日20:30:54。
通过debug发现产生问题的代码如下：
int len = 0;
while((len=socketChannel.read(buf))!=-1){
byte[] receData = buf.array();
String msg = new String(receData).trim();
System.out.println("接收来自客户端的数据为："+msg);
buf.clear();
}
发现在上面的while循环中是一个死循环，是不能够退出的。也就是说，socketChannel管道中的数据并不是读一次这个数据就没有了，而是还存于管道中，即使服务器端只发送一次数据给客户端，客户端通过如上的while循环的代码来读取数据是不会结束的。
将上面的代码更改为如下的代码即可：即采用更大的空间只读一次
//接收来自于客户端发送过来的数据
ByteBuffer buf = ByteBuffer.allocate(128);
socketChannel.read(buf);
byte[] receData = buf.array();
String msg = new String(receData).trim();
System.out.println("接收来自客户端的数据为："+msg);
buf.clear();
修正这个问题之后基本就客户端和服务器端基本就能够交互了
但是由于我这个Demo想完成的任务为：客户端和服务器端建立连接后，客户端第一次发送 1，第二个发送2，第N次发送N；服务器端每接收一个消息就给客户端发送一个“收到消息”。
但是发现服务器端收到的消息并不是1，2,3.。。。
查找原因发现客户端发送数据不能使用如下的代码：
//发送数据
buf.put((data+"").getBytes());//data初始值为1
while(buf.hasRemaining()){
socketChannel.write(buf);
}
data++;
换成如下的代码就可以正常工作了
socketChannel.write(ByteBuffer.wrap(new String(data+"").getBytes()));
data++;
至于原因，目前还不知道为什么也。
能够正常工作的完整代码可以在这里提取：https://github.com/wojiushimogui/Selector
小结
对Selector不怎么熟悉，模仿别人的代码来实现客户端和服务器端之间的交互，还存在一些问题。
Java 面试宝典是大明哥全力打造的 Java 精品面试题，它是一份靠谱、强大、详细、经典的 Java 后端面试宝典。它不仅仅只是一道道面试题，而是一套完整的 Java 知识体系，一套你 Java 知识点的扫盲贴。
它的内容包括：
- 大厂真题：Java 面试宝典里面的题目都是最近几年的高频的大厂面试真题。
- 原创内容：Java 面试宝典内容全部都是大明哥原创，内容全面且通俗易懂，回答部分可以直接作为面试回答内容。
- 持续更新：一次购买，永久有效。大明哥会持续更新 3+ 年，累计更新 1000+，宝典会不断迭代更新，保证最新、最全面。
- 覆盖全面：本宝典累计更新 1000+，从 Java 入门到 Java 架构的高频面试题，实现 360° 全覆盖。
- 不止面试：内容包含面试题解析、内容详解、知识扩展，它不仅仅只是一份面试题，更是一套完整的 Java 知识体系。
- 宝典详情：https://www.yuque.com/chenssy/sike-java/xvlo920axlp7sf4k
- 宝典总览：https://www.yuque.com/chenssy/sike-java/yogsehzntzgp4ly1
- 宝典进展：https://www.yuque.com/chenssy/sike-java/en9ned7loo47z5aw
目前 Java 面试宝典累计更新 400+ 道，总字数 42w+。大明哥还在持续更新中，下图是大明哥在 2024-12 月份的更新情况：
想了解详情的小伙伴，扫描下面二维码加大明哥微信【daming091】咨询
同时，大明哥也整理一套目前市面最常见的热点面试题。微信搜[大明哥聊 Java]或扫描下方二维码关注大明哥的原创公众号[大明哥聊 Java] ，回复【面试题】 即可免费领取。
