网络编程之Selector & SelectionKey详解(二)

这篇博文我希望能总结出一个实用的简单客户端服务器通信示例，我在前面几篇博文中说过，socket通信中存在数据无边界的问题，这样造成不知道read数据是否完成；在前面几篇博文中我们看到在对客户端SocketChannel进行注册感兴趣的事件时，并没有注册SelectionKey.OP_WRITE写事件，是因为写事件注册之后客户端通道大多数情况下都会立即满足可以写数据的条件，即使这个时候服务器逻辑还没有生成要写回去的数据，所以对于客户端SocketChannel的写事件，我们一般不是在注册通道时注册，而是在服务器准备好要写回的数据之后，通过选择键的interestOps(int
paramInt)这个方法进行设置。

这篇博文代码示例期望能结界，(1)数据无边界传输问题,(2)服务器端读写事件的转换问题。如果读者曾经看过Thrit通信组件的源码的话，你会发现笔者所总结的这个示例就是按照Thrit的服务器端通信模型进行编写的。我们下面直接上代码。

下面是客户端的代码，这个与前几篇博文中所使用的客户端通信代码差不多，只不过我这里使用的是阻塞模式的客户端通信模式。

package com.yujie.client;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
/**
* @author yujie.wang
*	SocketChannel 客户端代码测试
*/
public class SocketChannel_Client {

private final static String DEFAULT_HOST = "127.0.0.1";

private final static int DEFAULT_PORT = 3456;

private SocketChannel channel;

private Socket socket;

//分配一个大小为50字节的缓冲区 用于客户端通道的读写
private ByteBuffer buffer = ByteBuffer.allocate(50);

public SocketChannel_Client(){
this(DEFAULT_HOST, DEFAULT_PORT);
}

public SocketChannel_Client(String host, int port){
init(host,port);
}

/**
* 打开通道并设置对等的客户端socket对象
* 建立与服务端通道的连接
* @param host
* @param port
*/
public void init(String host, int port){
try {
//打开一个客户端通道，同时当前通道并没有与服务端通道建立连接
channel = SocketChannel.open();
//获得对等的客户端socket
socket = channel.socket();
//配置客户端socket
setSocket();
//将通道设置为非阻塞工作方式,如果通道设置我阻塞模式，那么这与直接使用传统Socket通信是一样的
//channel.configureBlocking(false);
//异步连接，发起连接之后就立即返回
//返回true，连接已经建立
//返回false，后续继续建立连接
channel.connect(new InetSocketAddress(host,port));
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
/**
* 验证连接是否建立
*/
public void finishConnect(){
try {
while(!channel.finishConnect()){
// nothing to do,wait connect
}
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

/**
* 验证当前连接是否可用
*/
public void isConnected(){
try {
if(channel == null || !channel.isConnected())
throw new IOException("channel is broken");
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

/**
* 配置客户端通道对等的Socket
*/
public void setSocket(){
try {
if(socket != null ){
//设置socket 读取的超时时间5秒
//socket.setSoTimeout(5000);
//设置小数据包不再组合成大包发送，也不再等待前一个数据包返回确认消息
socket.setTcpNoDelay(true);
//设置如果客户端Socket关闭了，未发送的包直接丢弃
socket.setSoLinger(true, 0);
}
} catch (Exception e) {
// TODO: handle exception
}
}
//java.lang.IllegalArgumentException
public void write(String data) {
byte [] datas = data.getBytes();
buffer.clear();
buffer.putInt(datas.length);
buffer.put(data.getBytes());
buffer.flip();
try {
// write并不一定能一次将buffer中的数据都写入 所以这里要多次写入
// 当多个线程同时调用同一个通道的写方法时，只有一个线程能工作，其他现在则会阻塞
while(buffer.hasRemaining()){
channel.write(buffer);
}
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

public void read(){
try {
buffer.clear();
// read方法并不阻塞,如果有数据读入返回读入的字节数，没有数据读入返回0 ，遇到流的末尾返回-1
// 当然这里和Socket和ServerSocket通信一样 也会存在消息无边界的问题 我们这里就采取简单的读取一次作为示例
System.out.println("read begin");
channel.read(buffer);
/*	while(buffer.hasRemaining() && channel.read(buffer) != -1){
printBuffer(buffer);
}*/
buffer.flip();
printBuffer(buffer);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

/**
* 输出buffer中的数据
* @param buffer
*/
public void printBuffer(ByteBuffer buffer){
while(buffer.hasRemaining()){
System.out.print((char)buffer.get());
}
System.out.println("");
System.out.println("****** Read end ******");
}

/**
* 判断通道是否打开
* @return
*/
public boolean isChannelOpen(){
try {
return channel.finishConnect() ? true : false;
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return false;
}

/**
* 关闭通道
*/
public void closeChannel(){
if(channel != null){
try {
channel.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

public static void main(String[] args) {
// TODO Auto-generated method stub
//	client(DEFAULT_HOST,DEFAULT_PORT);
SocketChannel_Client client = new SocketChannel_Client();
client.finishConnect();
System.out.println("connect success");
//这里连续进行三次读写，希望模拟的是连续三次客户端方法调用
for(int i = 0;i<3;i++){
client.write("Hello World");
client.read();
}
//客户端等待一段时间之后 直接退出
sleep(15000);
System.out.println("client exit");

}

public static void sleep(long time){
try {
Thread.sleep(time);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

}

服务器端代码我新起了一个线程用于处理SocketChannel的通信和业务处理逻辑：

package com.yujie.client;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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.util.Iterator;
import java.util.Set;
/**
* @author yujie.wang
*服务端代码示例，我单独起了一个线程用于处理SocketChannel通信和业务处理逻辑
*/
public class ServerSocketChannel_Server_Thread {

private ServerSocketChannel serverChannel;

private ServerSocket serverSocket;

private Selector selector;

private Thread thread;

private static int DEFAULT_BIND_PORT = 3456;

public ServerSocketChannel_Server_Thread(){
this(DEFAULT_BIND_PORT);
}

public ServerSocketChannel_Server_Thread(int bindPort){
init(bindPort);
}

public void init(int bindPort){
try {
//打开一个服务端通道
serverChannel = ServerSocketChannel.open();
//获得对等的ServerSocket对象
serverSocket = serverChannel.socket();
//将服务端ServerSocket绑定到指定端口
serverSocket.bind(new InetSocketAddress(bindPort));
System.out.println("Server listening on port: "+ bindPort);
//将通道设置为非阻塞模式
serverChannel.configureBlocking(false);
//打开一个选择器
selector = Selector.open();
//将通道注册到打开的选择器上
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
} catch (IOException e) {
// TODO Auto-generated catch block
System.out.println("init exception: "+e);
}

}

public void select(){
try {
//select()方法会阻塞，直到有准备就绪的通道有准备好的操作;或者当前线程中断该方法也会返回
//这里的返回值不是选择器中已选择键集合中键的数量，而是从上一次select()方法调用到这次调用期间进入就绪状态通道的数量
selector.select();
//获得已选择键的集合这个集合中包含了 新准备就绪的通道和上次调用select()方法已经存在的就绪通道
Set<SelectionKey> set = selector.selectedKeys();
Iterator it = set.iterator();
while(it.hasNext()){
SelectionKey key = (SelectionKey)it.next();
//通过调用remove将这个键key从已选择键的集合中删除
it.remove();
if(key.isAcceptable
e0c4
()){
handleAccept(key);
}else if(key.isReadable()){
handleRead(key);
}else if(key.isWritable()){
handleWrite(key);
}

}
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

public void handleAccept(SelectionKey key){
try {
//因为能注册SelectionKey.OP_ACCEPT事件的只有 ServerSocketChannel通道，
//所以这里可以直接转换成ServerSocketChannel
ServerSocketChannel channel = (ServerSocketChannel)key.channel();
//获得客户端的SocketChannel对象 ,accept这里不会阻塞，如果没有连接到来，这里会返回null
SocketChannel clientChannel = channel.accept();
System.out.println("Accepted Connected from: "+ clientChannel);
//将客户端socketChannel设置为非阻塞模式
clientChannel.configureBlocking(false);
//为该客户端socket分配一个ByteBuffer
SelectionKey clientKey = clientChannel.register(selector, SelectionKey.OP_READ);
BufferFrame buffer = new BufferFrame(clientChannel,clientKey);
clientKey.attach(buffer);
} catch (IOException e) {
// TODO Auto-generated catch block
System.out.println("handleAccept exception: "+ e);
}
}

public void handleRead(SelectionKey key){
//从键中获得相应的客户端socketChannel
BufferFrame buffer = (BufferFrame)key.attachment();
//从通道中读取数据 如果读取数据失败 则关闭通道 注销选择键中通道和缓冲区之间的注册关系
if(!buffer.read()){
System.out.println("buffer read error");
cleanSelectionkey(key);
}
//如果通道读取完数据之后 处理数据 并改变通道所注册的缓冲区的状态
if(buffer.readCompleteFrame()){
//System.out.println("print data");
buffer.printReadContent();
}
}

/**
* 通过通道将数据写会调用请求的客户端
* @param key
*/
public void handleWrite(SelectionKey key){
//System.out.println("handle write");
BufferFrame buffer = (BufferFrame)key.attachment();
if(!buffer.write())
cleanSelectionkey(key);
}

/**
* 关闭通道 取消通道和选择器之间的注册关系
* @param key
*/
public void cleanSelectionkey(SelectionKey key){
try {
BufferFrame buffer = (BufferFrame)key.attachment();
key.channel().close();
key.cancel();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

public void serverStart(){
//开启一个线程
thread = new Thread( new AcceptImplThread());
thread.start();
}

/**
* 该方法时期望主线程阻塞，指导任务线程执行结束
*/
public void joinThread(){
try {
thread.join();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

class AcceptImplThread implements Runnable{

@Override
public void run() {
// TODO Auto-generated method stub
try {
while(true){
System.out.println("select begin to run");
select();
System.out.println("select end to run");
}
} catch (Exception e) {
// TODO: handle exception
System.out.println("run exception: "+e);
}finally{
try {
for(SelectionKey key : selector.keys()){
key.channel().close();
key.cancel();
}
} catch (Exception e2) {
// TODO: handle exception
}
}
}

}

public static void main(String[] args) {
// TODO Auto-generated method stub
ServerSocketChannel_Server_Thread server = new ServerSocketChannel_Server_Thread();
server.serverStart();
server.joinThread();
System.out.println("server exit");
}

}


BufferFrame类主要是完成对客户端发送数据的读写转换，以及通道自身读写事件的注册转换
package com.yujie.client;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;

/**
* @author yujie.wang
* 该方法用于读写客户端socketChannel
* 这个类主要解决就是服务端数据的读写转换和读写事件的注册
*/
public class BufferFrame {
private SocketChannel clientChannel;

private SelectionKey key;

private ByteBuffer buf;

private BufferFrameState state = BufferFrameState.READING_FRAME_SIZE;

public BufferFrame(SocketChannel channel, SelectionKey key){
this.clientChannel = channel;
this.key = key;
buf = ByteBuffer.allocate(4);
}

public boolean read(){
//读取数据包头部 获取待读取数据的大小
if(state == BufferFrameState.READING_FRAME_SIZE){
if(!internalRead())
return false;
}
// 读取完包的大小之后 取出数据大小 并为缓冲区分配大小，改变当前读取数据的状态为BufferFrameState.READING_FRAME
if(buf.remaining() == 0){
//System.out.println("read fram size end");
int frameSize = buf.getInt(0);
System.out.println("frameSize: "+ frameSize);
if(frameSize < 0)
return false;
buf = ByteBuffer.allocate(frameSize);
state = BufferFrameState.READING_FRAME;
}else {
return true;
}
//读取客户端发过来的数据
if(state == BufferFrameState.READING_FRAME){
if(!internalRead())
return false;
//读取完数据之后清空该选择键所感兴趣的事件，并改变读取数据的状态为READ_FRAME_COMPLETE
if(buf.remaining() == 0){
//System.out.println("read fram end");
key.interestOps(0);
state = BufferFrameState.READ_FRAME_COMPLETE;
return true;
}
}
System.out.println("read error");
return false;

}

/**
* 从通道中读取数据到buf中，每一次都从内核中尽量多的读取数据尝试填满buf，当然如果一次填不满buf数据，要进行多次读取
* 当然如果没有数据读取 在通道处于非阻塞模式下改方法不会阻塞
* @return
*/
public boolean internalRead(){
try {
if(clientChannel.read(buf) < 0){
System.out.println("read error inter");
return false;
}
return true;
} catch (IOException e) {
// TODO Auto-generated catch block
System.out.println("BufferFrame read exception: "+ e);
return false;
}
}

/**
* 将数据写回调用方
* @return
*/
public boolean write(){
if(state == BufferFrameState.WRITING){
try {
//System.out.println("write begin");
if(clientChannel.write(buf)<0)
return false;
} catch (IOException e) {
// TODO Auto-generated catch block
System.out.println("write exception: "+e);
return false;
}

if(buf.remaining() == 0){
System.out.println("write end");
key.interestOps(SelectionKey.OP_READ);
buf = ByteBuffer.allocate(4);
state = BufferFrameState.READING_FRAME_SIZE;
return true;
}
}
return false;

}

/**
* 改变当前缓冲区或者通道的状态
*/
public void changeKeyState(){
try {
if(state == BufferFrameState.AWAITING_REGISTER_WRITE){
//System.out.println("a");
key.interestOps(SelectionKey.OP_WRITE);
//System.out.println("b");
state = BufferFrameState.WRITING;
}else if(state == BufferFrameState.AWAITING_REGISTER_READ){
key.interestOps(SelectionKey.OP_READ);
buf = ByteBuffer.allocate(4);
state = BufferFrameState.READING_FRAME_SIZE;
} else if(state ==BufferFrameState.AWAITING_CLOSE){
clientChannel.close();
key.cancel();
} else{
System.out.println("changeSelectInterest was called, but state is invalid (" + state + ")");
}
} catch (Exception e) {
// TODO: handle exception
System.out.println("changeKeyState exception: "+e);
}
}

public void printReadContent(){
buf.flip();
while(buf.hasRemaining()){
//System.out.println("positon: " + buf.position()+ " limit:"+ buf.limit());
System.out.print((char)buf.get());
}
buf.clear();
System.out.println("");
System.out.println("****** Read end ******");
System.out.println("");

//处理完数据之后 将当前缓冲区改为写数据状态
buf.put("ServerData".getBytes());
buf.flip();
//System.out.println("ready write data");
state = BufferFrameState.AWAITING_REGISTER_WRITE;
changeKeyState();
}

public boolean readCompleteFrame(){
return state == BufferFrameState.READ_FRAME_COMPLETE;
}

/**
* 定义枚举用于标示当前数据流的读取状态
* 这些状态的实际转换 要通过SelectionKey.interestOps(int ops)方法进行转换
*/
private enum BufferFrameState {
//标示客户端正在读取数据大小 客户端每次写入数据都是先写入 要读取的数据大小
READING_FRAME_SIZE,
//标示客户端正在读取数据
READING_FRAME,
//标示客户端读取数据完成。可以向写入数据转换了
READ_FRAME_COMPLETE,
//当客户端读取完数据之后，客户端socketChannel可以注册写事件
AWAITING_REGISTER_WRITE,
//标示正在写数据
WRITING,
//写完数据之后 可以向读取数据转换
AWAITING_REGISTER_READ,
//标示当前通道可以进行关闭
AWAITING_CLOSE
}
}


如果正确执行客户端和服务端代码客户端会输出如下结果：
connect success
read begin
ServerData
****** Read end ******
read begin
ServerData
****** Read end ******
read begin
ServerData
****** Read end ******

服务端会输出如下结果：
Server listening on port: 3456
select begin to run
Accepted Connected from: java.nio.channels.SocketChannel[connected local=/127.0.0.1:3456 remote=/127.0.0.1:56291]
select end to run
select begin to run
frameSize: 11
Hello World
****** Read end ******

select end to run
select begin to run
write end
select end to run
select begin to run
frameSize: 11
Hello World
****** Read end ******

select end to run
select begin to run
write end
select end to run
select begin to run
frameSize: 11
Hello World
****** Read end ******

select end to run
select begin to run
write end
select end to run
select begin to run