hadoop的源码阅读，ipc包学习--nio

在看hadoop的源码，看到ipc包，涉及到nio和rpc的知识。为了加深了解，把思考的过程写下来。

一、nio是java的非阻塞式通信方式，区别于我们常见的socket通信方式（阻塞式）。

比如说传统的socket请求是这样的，客户端发送一个请求给服务器，服务器创建一个线程去处理这个请求，可能这个请求的业务过程比较长，那么两边的网络连接是一直连接状态的，直到服务端的处理完请求，才断开网络连接。

在nio模式下，客户端和服务器端都会建立一个单独的通信线程去管理所有的数据传输请求。这样的话客户端业务线程只要将要输入的数据传入到这个通信线程里面，就可以进入等待状态（线程wait（））不用占用线程资源。而服务端的业务线程从服务端的通信线程中获取到业务请求并处理后返回结果，同样只需要将结果返回给通信线程然后进入等待下一个请求的阶段，由通信线程去完成数据的传输。而两边的通信线程之所以能够管理如此多的异步通信，主要有两点：buffer和select。每个通信链接都会有一个对应的数据存储区（buffer），客户端业务线程将数据写入到buffer，由通信线程将数据发送出去。服务器端的通信线程接收到数据，并将数据写入到buffer，并触（select中注册的事件）发服务端的业务线程读取。nio没有数据结束的标志，所以，在ipc传输中，根据设定好的对象类型来确定返回结果的长度。

以下是网上找的例子，稍有修改。NIO服务端：

package org.czc.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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 int flag = 0;
/* 缓冲区大小 */
private int BLOCK = 4096;
/* 接受数据缓冲区 */
private ByteBuffer sendbuffer = ByteBuffer.allocate(BLOCK);
/* 发送数据缓冲区 */
private ByteBuffer receivebuffer = ByteBuffer.allocate(BLOCK);
private Selector selector;

public NIOServer(int port) throws IOException {
// 打开服务器套接字通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 服务器配置为非阻塞
serverSocketChannel.configureBlocking(false);
// 检索与此通道关联的服务器套接字
ServerSocket serverSocket = serverSocketChannel.socket();
// 进行服务的绑定
serverSocket.bind(new InetSocketAddress(port));
// 通过open()方法找到Selector
selector = Selector.open();
// 注册到selector，等待连接
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("Server Start----8888:");
}

// 监听
private void listen() throws IOException {
while (true) {
// 选择一组键，并且相应的通道已经打开
selector.select();
// 返回此选择器的已选择键集。
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey selectionKey = iterator.next();
iterator.remove();
handleKey(selectionKey);
}
}
}

// 处理请求
private void handleKey(SelectionKey selectionKey) throws IOException {
// 接受请求
ServerSocketChannel server = null;
SocketChannel client = null;
String receiveText;
String sendText;
int count = 0;
// 测试此键的通道是否已准备好接受新的套接字连接。
if (selectionKey.isAcceptable()) {
// 返回为之创建此键的通道。
server = (ServerSocketChannel) selectionKey.channel();
// 接受到此通道套接字的连接。
// 此方法返回的套接字通道（如果有）将处于阻塞模式。
client = server.accept();
// 配置为非阻塞
client.configureBlocking(false);
// 注册到selector，等待连接
client.register(selector, SelectionKey.OP_READ);
} else if (selectionKey.isReadable()) {
// 返回为之创建此键的通道。
client = (SocketChannel) selectionKey.channel();
// 将缓冲区清空以备下次读取
receivebuffer.clear();
// 读取服务器发送来的数据到缓冲区中
count = client.read(receivebuffer);
if (count > 0) {
receiveText = new String(receivebuffer.array(), 0, count);
System.out.println("服务器端接受客户端数据--:" + receiveText);
client.register(selector, SelectionKey.OP_WRITE);
}
} else if (selectionKey.isWritable()) {
// 将缓冲区清空以备下次写入
sendbuffer.clear();
// 返回为之创建此键的通道。
client = (SocketChannel) selectionKey.channel();
sendText = "message from server--" + flag++;
// 向缓冲区中输入数据
sendbuffer.put(sendText.getBytes());
// 将缓冲区各标志复位,因为向里面put了数据标志被改变要想从中读取数据发向服务器,就要复位
sendbuffer.flip();
// 输出到通道
client.write(sendbuffer);
System.out.println("服务器端向客户端发送数据--：" + sendText);
client.register(selector, SelectionKey.OP_READ);
}
}

/**
* @param args
* @throws IOException
*/
public static void main(String[] args) throws IOException {
// TODO Auto-generated method stub
int port = 8888;
NIOServer server = new NIOServer(port);
server.listen();
}
}

NIO写的客户端：

package org.czc.mytest;

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;

/**
* NIO客户端
* @author 小路
*/
public class NIOClient {
//通道管理器
private Selector selector;

/**
* 获得一个Socket通道，并对该通道做一些初始化的工作
* @param ip 连接的服务器的ip
* @param port  连接的服务器的端口号
* @throws IOException
*/
public void initClient(String ip,int port) throws IOException {
// 获得一个Socket通道
SocketChannel 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) {
selector.select();
// 获得selector中选中的项的迭代器
Iterator ite = this.selector.selectedKeys().iterator();
while (ite.hasNext()) {
SelectionKey key = (SelectionKey) ite.next();
// 删除已选的key,以防重复处理
ite.remove();
// 连接事件发生
if (key.isConnectable()) {
SocketChannel channel = (SocketChannel) key
.channel();
// 如果正在连接，则完成连接
if(channel.isConnectionPending()){
channel.finishConnect();

}
// 设置成非阻塞
channel.configureBlocking(false);

//在这里可以给服务端发送信息哦
channel.write(ByteBuffer.wrap(new String("向服务端发送了一条信息").getBytes()));
//在和服务端连接成功之后，为了可以接收到服务端的信息，需要给通道设置读的权限。
channel.register(this.selector, SelectionKey.OP_READ,null);

// 获得了可读的事件
} else if (key.isReadable()) {
read(key);
}

}

}
}
/**
* 处理读取服务端发来的信息 的事件
* @param key
* @throws IOException
*/
public void read(SelectionKey key) throws IOException{
//和服务端的read方法一样
}

/**
* 启动客户端测试
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NIOClient client = new NIOClient();
client.initClient("localhost",8000);
client.listen();
}

}

普通socket访问nio接口

package org.czc.mytest;

import java.io.BufferedInputStream;
import java.io.BufferedWriter;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.Socket;

public class SocketClient {
/**
* 方法名：main 描述： 作者：白鹏飞 日期：2012-8-23 下午01:47:12
*
* @param @param args
* @return void
*/
public static void main(String[] args) {
Socket socket = null;
DataInputStream br = null;
PrintWriter pw = null;
try {
//客户端socket指定服务器的地址和端口号
socket = new Socket("127.0.0.1", 8888);
socket.setTcpNoDelay(true);
System.out.println("Socket=" + socket);
//同服务器原理一样
br = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
//            		new BufferedReader(new InputStreamReader(
//                    socket.getInputStream()));
pw = new PrintWriter(new BufferedWriter(new OutputStreamWriter(
socket.getOutputStream())));
for (int i = 0; i < 10; i++) {
pw.println("howdy " + i);
pw.flush();
byte[] buffer = new byte[1024];
int bytesRead = 0;
do{
bytesRead = br.read(buffer);
System.out.println(new String(buffer));
}while(bytesRead==1024);
}
pw.println("END");
pw.flush();
Thread.sleep(100);
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
System.out.println("close......");
br.close();
pw.close();
socket.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

}

第一次写这么长的技术博客，感觉很不满意，但还是比较有收获的。

至少发现，你看完一篇文章，自以为理解了，但是让你自己再来讲一遍，你才会发现，其实你理解的很肤浅。