Java NIO系列教程（六） 多路复用器Selector

一、为什么使用Selector?

二、Selector简介

三、Selector的使用

3.1、Selector的创建

Selector selector = Selector.open();

3.2、将channel注册到Selector中

channel.configureBlocking(false);
SelectionKey key = channel.register(selector, Selectionkey.OP_READ);

int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;

SelectionKey

interest集合

int interestSet = selectionKey.interestOps();

boolean isInterestedInAccept  = (interestSet & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT；
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;

ready集合

int readySet = selectionKey.readyOps();

selectionKey.isAcceptable();//等价于selectionKey.readyOps()&SelectionKey.OP_ACCEPT
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();

取出SelectionKey所关联的Selector和Channel

Channel  channel  = selectionKey.channel();
Selector selector = selectionKey.selector();

关于取消SelectionKey对象

cancel()

select()

为SelectionKey绑定附加对象

selectionKey.attach(theObject);
Object attachedObj = selectionKey.attachment();

SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

isRegistered（）

3.3、通过Selector选择通道

注意

selectedKeys()

Set selectedKeys = selector.selectedKeys();

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()" 将这个 key 从迭代器中删除, 因为 select() 方法仅仅是简单地将就绪的 IO 操作放到 selectedKeys 集合中, 因此如果我们从 selectedKeys 获取到一个 key, 但是没有将它删除, 那么下一次 select 时, 这个 key 所对应的 IO 事件还在 selectedKeys 中.

注意, 我们可以动态更改 SekectedKeys 中的 key 的 interest set.

关于Selector执行选择的过程

select（）

select（）

select（）

cancle()

select（）

select（）

select()

深入已注册键集合的管理

keyIterator.remove()

停止选择

close（）

wakeup（）

interrupt()

wakeup()

Selector 的基本使用流程

如果是 OP_ACCEPT 事件, 则调用 "SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept()" 获取 SocketChannel, 并将它设置为 非阻塞的, 然后将这个 Channel 注册到 Selector 中.

完整的示例

package com.dxz.springsession.nio.demo4;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

public class ServerSocketChannelTest {

private int size = 1024;
private ServerSocketChannel socketChannel;
private ByteBuffer byteBuffer;
private Selector selector;
private final int port = 8998;
private int remoteClientNum=0;

public ServerSocketChannelTest() {
try {
initChannel();
} catch (Exception e) {
e.printStackTrace();
System.exit(-1);
}
}

public void initChannel() throws Exception {
socketChannel = ServerSocketChannel.open();
socketChannel.configureBlocking(false);
socketChannel.bind(new InetSocketAddress(port));
System.out.println("listener on port:" + port);
selector = Selector.open();
socketChannel.register(selector, SelectionKey.OP_ACCEPT);
byteBuffer = ByteBuffer.allocateDirect(size);
byteBuffer.order(ByteOrder.BIG_ENDIAN);
}

private void listener() throws Exception {
while (true) {
int n = selector.select();
if (n == 0) {
continue;
}
Iterator<SelectionKey> ite = selector.selectedKeys().iterator();
while (ite.hasNext()) {
SelectionKey key = ite.next();
//a connection was accepted by a ServerSocketChannel.
if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel channel = server.accept();
registerChannel(selector, channel, SelectionKey.OP_READ);
remoteClientNum++;
System.out.println("online client num="+remoteClientNum);
replyClient(channel);
}
//a channel is ready for reading
if (key.isReadable()) {
readDataFromSocket(key);
}

ite.remove();//must
}

}
}

protected void readDataFromSocket(SelectionKey key) throws Exception {
SocketChannel socketChannel = (SocketChannel) key.channel();
int count;
byteBuffer.clear();
while ((count = socketChannel.read(byteBuffer)) > 0) {
byteBuffer.flip(); // Make buffer readable
// Send the data; don't assume it goes all at once
while (byteBuffer.hasRemaining()) {
socketChannel.write(byteBuffer);
}
byteBuffer.clear(); // Empty buffer
}
if (count < 0) {
socketChannel.close();
}
}

private void replyClient(SocketChannel channel) throws IOException {
byteBuffer.clear();
byteBuffer.put("hello client!\r\n".getBytes());
byteBuffer.flip();
channel.write(byteBuffer);
}

private void registerChannel(Selector selector, SocketChannel channel, int ops) throws Exception {
if (channel == null) {
return;
}
channel.configureBlocking(false);
channel.register(selector, ops);
}

public static void main(String[] args) {
try {
new ServerSocketChannelTest().listener();
} catch (Exception e) {
e.printStackTrace();
}
}
}

//客户端
package com.dxz.springsession.nio.demo4;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.SocketChannel;

public class SocketChannelTest {

private int size = 1024;
private ByteBuffer byteBuffer;
private SocketChannel socketChannel;

public void connectServer() throws IOException {
socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("127.0.0.1", 8998));
byteBuffer = ByteBuffer.allocate(size);
byteBuffer.order(ByteOrder.BIG_ENDIAN);
receive();
}

private void receive() throws IOException {
while (true) {
int count;
byteBuffer.clear();
while ((count = socketChannel.read(byteBuffer)) > 0) {
byteBuffer.flip();
while (byteBuffer.hasRemaining()) {
System.out.print((char) byteBuffer.get());
}
//send("send data to server\r\n".getBytes());
byteBuffer.clear();
}
}
}

private void send(byte[] data) throws IOException {
byteBuffer.clear();
byteBuffer.put(data);
byteBuffer.flip();
socketChannel.write(byteBuffer);
}

public static void main(String[] args) throws IOException {
new SocketChannelTest().connectServer();
}
}