Java之NIO(二)selector socketChannel
浏览数：35 /
时间：2015年06月08日
上篇文章对NIO进行了简介，对Channel和Buffer接口的使用进行了说明，并举了一个简单的例子来说明其使用方法。
本篇则重点说明selector，Selector（选择器）是Java NIO中能够检测一到多个NIO通道，并能够知晓通道是否为诸如读写事件做好准备的组件。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接。
与selector联系紧密的是ServerSocketChannel和SocketChannel，他们的使用与上篇文章描述的FileChannel的使用方法类似，然后与ServerSocket和Socket也有一些联系。
本篇首先简单的进selector进行说明，然后一个简单的示例程序，来演示即时通讯。
Selector
使用传统IO进行网络编程，如下图所示：
每一个到服务端的连接，都需要一个单独的线程（或者线程池）来处理其对应的socket，当连接数多的时候，对服务端的压力极大。并使用socket的getInputStream。Read方法来不断的轮训每个socket，效率可想而知。
而selector则可以在同一个线程中监听多个channel的状态，当某个channel有selector感兴趣的事情发现，selector则被激活。即不会主动去轮询。如下图所示：
Selector使用如下示意：
public static void main(String[] args) throws IOException {
Selector selector = Selector.open();//声明selector
ServerSocketChannel sc = ServerSocketChannel.open();
sc.configureBlocking(false);//必须设置为异步
sc.socket().bind(new InetSocketAddress(8081));//绑定端口
//把channel 注册到 selector上
sc.register(selector, SelectionKey.OP_ACCEPT|SelectionKey.OP_CONNECT|SelectionKey.OP_READ|SelectionKey.OP_WRITE);
while(true){
selector.select();//阻塞，直到注册的channel上某个感兴趣的事情发生
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> 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();
}
}
}极简即时通讯
本例子是是一个极为简单的例子，很多地方都不完善，但是例子可以很好的说明selector的使用方法。
本例子包含服务端和客户端两个部分，其中服务端采用两个selector，用来建立连接和数据的读写。两个selector在两个线程中。
服务端
/**
* 简单的即时通讯服务端，采用建立连接 selector和数据 selector分离。很不完善
*
*/
public class ServerSocketChannelTest {
private static final int SERVER_PORT = 8081;
private ServerSocketChannel server;
private volatile Boolean isStop = false;
//负责建立连接的selector
private Selector conn_Sel;
//负责数据读写的selector
private Selector read_Sel;
// private ExecutorService sendService = Executors.newFixedThreadPool(3);
//锁，用来在建立连接后，唤醒read_Sel时使用的同步
private Object lock = new Object();
//注册的用户
private Map<String, ClientInfo> clents = new HashMap<String, ClientInfo>();
/**
* 初始化，绑定端口
*/
public void init() throws IOException {
//创建ServerSocketChannel
server = ServerSocketChannel.open();
//绑定端口
server.socket().bind(new InetSocketAddress(SERVER_PORT));
server.configureBlocking(false);
//定义两个selector
conn_Sel = Selector.open();
read_Sel = Selector.open();
//把channel注册到selector上，第二个参数为兴趣的事件
server.register(conn_Sel, SelectionKey.OP_ACCEPT);
}
// 负责建立连接。
private void beginListen() {
System.out.println("--------开始监听----------");
while (!isStop) {
try {
conn_Sel.select();
} catch (IOException e) {
e.printStackTrace();
continue;
}
Iterator<SelectionKey> it = conn_Sel.selectedKeys().iterator();
while (it.hasNext()) {
SelectionKey con = it.next();
it.remove();
if (con.isAcceptable()) {
try {
SocketChannel newConn = ((ServerSocketChannel) con
.channel()).accept();
handdleNewInConn(newConn);
} catch (IOException e) {
e.printStackTrace();
continue;
}
} else if (con.isReadable()) {//废代码，执行不到。
try {
handleData((SocketChannel) con.channel());
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
/**
* 负责接收数据
*/
private void beginReceive(){
System.out.println("---------begin receiver data-------");
while (true) {
synchronized (lock) {
}
try {
read_Sel.select();
} catch (IOException e) {
e.printStackTrace();
continue;
}
Iterator<SelectionKey> it = read_Sel.selectedKeys().iterator();
while (it.hasNext()) {
SelectionKey con = it.next();
it.remove();
if (con.isReadable()) {
try {
handleData((SocketChannel) con.channel());
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
private void handdleNewInConn(SocketChannel newConn) throws IOException {
newConn.configureBlocking(false);
//这里必须先唤醒read_Sel，然后加锁，防止读写线程的中select方法再次锁定。
synchronized (lock) {
read_Sel.wakeup();
newConn.register(read_Sel, SelectionKey.OP_READ);
}
//newConn.register(conn_Sel, SelectionKey.OP_READ);
}
private void handleData(final SocketChannel data) throws IOException {
ByteBuffer buffer = ByteBuffer.allocate(512);
try {
int size= data.read(buffer);
if (size==-1) {
System.out.println("-------连接断开-----");
//这里暂时不处理，这里可以移除已经注册的客户端
}
} catch (IOException e) {
e.printStackTrace();
return;
}
buffer.flip();
byte[] msgByte = new byte[buffer.limit()];
buffer.get(msgByte);
Message msg = Message.getMsg(new String(msgByte));
//这里读完数据其实已经可以另开线程了下一步的处理，理想情况下，根据不同的消息类型，建立不同的队列，把待发送的消息放进队列
//当然也可以持久化。如果在数据没有读取前，另开线程的话，读写线程中 read_Sel.select(),会立刻返回。可以把
if (msg.getType().equals("0")) {// 注册
ClientInfo info = new ClientInfo(msg.getFrom(), data);
clents.put(info.getClentID(), info);
System.out.println(msg.getFrom() + "注册成功");
} else {// 转发
System.out.println("收到"+msg.getFrom()+"发给"+msg.getTo()+"的消息");
ClientInfo to = clents.get(msg.getTo());
buffer.rewind();
if (to != null) {
SocketChannel sendChannel = to.getChannel();
try {
while (buffer.hasRemaining()) {
sendChannel.write(buffer);
}
} catch (Exception e) {
}
finally {
buffer.clear();
}
}
}
}
public static void main(String[] args) throws IOException {
final ServerSocketChannelTest a = new ServerSocketChannelTest();
a.init();
new Thread("receive..."){
public void run() {
a.beginReceive();
};
}.start();
a.beginListen();
}
}客户端
/**
* new 次对象，然后调用start方法，其中self 是自己id
*
* to 是接收人id
*
*/
public class Client {
/**
* 自己的ID
*/
private String self;
/**
* 接收人ID
*/
private String to;
//通道管理器
private Selector selector;
private ByteBuffer writeBuffer = ByteBuffer.allocate(512);
private SocketChannel channel;
private Object lock = new Object();
private volatile boolean isInit = false;
public Client(String self, String to) {
super();
this.self = self;
this.to = to;
}
/**
* 获得一个Socket通道，并对该通道做一些初始化的工作
* @param ip 连接的服务器的ip
* @param port 连接的服务器的端口号
* @throws IOException
*/
public void initClient(String ip,int port) throws IOException {
// 获得一个Socket通道
channel = SocketChannel.open();
// 设置通道为非阻塞
channel.configureBlocking(false);
// 获得一个通道管理器
this.selector = Selector.open();
// 客户端连接服务器,其实方法执行并没有实现连接，需要在listen（）方法中调
//用channel.finishConnect();才能完成连接
channel.connect(new InetSocketAddress(ip,port));
//将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_CONNECT事件。
channel.register(selector, SelectionKey.OP_CONNECT);
}
/**
* 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理
* @throws IOException
*/
@SuppressWarnings("unchecked")
public void listen() throws IOException {
// 轮询访问selector
while (true) {
synchronized (lock) {
}
selector.select();
// 获得selector中选中的项的迭代器
Iterator<SelectionKey> ite = this.selector.selectedKeys().iterator();
while (ite.hasNext()) {
SelectionKey key = ite.next();
// 删除已选的key,以防重复处理
ite.remove();
// 连接事件发生
if (key.isConnectable()) {
SocketChannel channel = (SocketChannel) key
.channel();
// 如果正在连接，则完成连接
if(channel.isConnectionPending()){
channel.finishConnect();
}
// 设置成非阻塞
channel.configureBlocking(false);
//在和服务端连接成功之后，为了可以接收到服务端的信息，需要给通道设置读的权限。
channel.register(this.selector, SelectionKey.OP_READ);
isInit = true;
// 获得了可读的事件
} else if (key.isReadable()) {
read(key);
}
}
}
}
/**
* 处理读取服务端发来的信息的事件
* @param key
* @throws IOException
*/
public void read(SelectionKey key) throws IOException{
SocketChannel data = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(512) ;
try {
data.read(buffer );
} catch (IOException e) {
e.printStackTrace();
data.close();
return;
}
buffer.flip();
byte[] msgByte = new byte[buffer.limit()];
buffer.get(msgByte);
Message msg = Message.getMsg(new String(msgByte));
System.out.println("---收到消息--"+msg+" 来自 "+msg.getFrom());
}
private void sendMsg(String content){
writeBuffer.put(content.getBytes());
writeBuffer.flip();
try {
while (writeBuffer.hasRemaining()) {
channel.write(writeBuffer);
}
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
writeBuffer.clear();
}
/**
* 启动客户端测试
* @throws IOException
*/
public void start() throws IOException {
initClient("localhost",8081);
new Thread("reading"){
public void run() {
try {
listen();
} catch (IOException e) {
e.printStackTrace();
}
};
}.start();
int time3 = 0;
while(!isInit&&time3<3){
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
time3 ++;
}
System.out.println("--------开始注册------");
Message re = new Message("", self, "");
sendMsg(re.toString());
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("-----注册成功----");
String content ="";
System.out.println("---- 请输入要发送的消息，按回车发送，输入 123 退出----------");
Scanner s = new Scanner(System.in);
while (!content.equals("123")&&s.hasNext()) {
content = s.next();
Message msg = new Message(content, self, to);
msg.setType("1");
sendMsg(msg.toString());
if (content.equals("123")) {
break;
}
System.out.println("---发送成功---");
}
channel.close();
}
}客户端测试
public class TestClient1 {
public static void main(String[] args) throws IOException {
Client c1 =new Client("1", "2");
c1.start();
}
}
public class TestClient2 {
public static void main(String[] args) throws IOException {
Client c2 =new Client("2", "1");
c2.start();
}
}结束
本文的例子极为简单，但是都经过测试。在编码的过程中，遇到的问题主要有两点：
1. channel.register()方法阻塞
2. 使用线程池遇到问题。本文最后在服务端的读写线程中，没有使用线程池，原因注释说的比较明白，也说明了使用线程池的一种设想。
另外在本文编码过程中，遇到了一些问题，去网上寻求答案，遇到了一些不错的文章，本文某些部分由参考。
selector的讲解，官方文档翻译 http://ifeve.com/selectors/
NIO就绪的OP_write http://blog.csdn.net/zhouhl_cn/article/details/6582435
此文不错：http://blog.csdn.net/jjzhk/article/details/39553613
另外还有两个反面教材：
郑重声明：本站内容如果来自互联网及其他传播媒体，其版权均属原媒体及文章作者所有。转载目的在于传递更多信息及用于网络分享，并不代表本站赞同其观点和对其真实性负责，也不构成任何其他建议。
