客户端使用Java的阻塞IO
服务端使用Java的非阻塞NIO
package com.nio.echo; import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Scanner; /**
* @author 作者 E-mail:
* @version 创建时间：2015-10-29 下午02:49:47 类说明
*/
public class EchoClient
{
public static final String REMOT_IP = "127.0.0.1"; public static final int REMOTE_PORT = 8080; public void connectServer() throws IOException
{
Socket socket = new Socket(); socket.connect(new InetSocketAddress(REMOT_IP, REMOTE_PORT)); if (socket.isConnected())
{
System.out.println("connect remote address success");
} // 启动线程监听server端消息
new Thread(new client2server(socket)).start();
Scanner scanner = new Scanner(System.in); OutputStream output = socket.getOutputStream();
while (true)
{
String str = scanner.nextLine(); if (str.equals("quit"))
{
socket.close();
break;
}
output.write(str.getBytes("UTF-8")); } } public static void main(String[] args) throws IOException
{
new EchoClient().connectServer();
}
} class client2server implements Runnable
{
private Socket socket = null; public client2server(Socket socket)
{
this.socket = socket;
} @Override
public void run()
{
InputStream inputStream;
try
{
inputStream = socket.getInputStream();
byte[] bytes = new byte[1024];
while (true)
{
int num = inputStream.read(bytes);
if (num != -1)
{
System.out.print(num + " ");
}
else
{
System.out.println("server is shutup");
break;
} String str = new String(bytes, 0, num, "UTF-8");
System.out.println("get data: " + str); }
}
catch(IOException e)
{
e.printStackTrace();
} }
}
package com.nio.echo; import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Iterator;
import java.util.Set; /**
* @author 作者 E-mail:
* @version 创建时间：2015-10-29 下午02:49:12 类说明
*/
public class NIOEchoServer
{
private static ServerSocketChannel ssc = null; private static Selector selector = null; private static final int PORT = 8080; public static void startServer() throws IOException
{
ssc = ServerSocketChannel.open();
selector = Selector.open();
ssc.configureBlocking(false); // nio 对socket 和serverSocket进行了怎样封装
final ServerSocket serverSocket = ssc.socket(); serverSocket.bind(new InetSocketAddress(PORT));
serverSocket.setReuseAddress(true); final AcceptHandler acceptHandler = new AcceptHandler();
ssc.register(selector, SelectionKey.OP_ACCEPT, acceptHandler);
while (true)
{
int n = selector.select();
if (n == 0)
continue; final Set<SelectionKey> readyKeys = selector.selectedKeys();
final Iterator<SelectionKey> it = readyKeys.iterator();
while (it.hasNext())
{
final SelectionKey key = it.next();
final Handle handler = (Handle) key.attachment();
handler.doHandle(key);
it.remove();
}
}
} public static void main(String[] args) throws IOException
{
NIOEchoServer.startServer();
}
} interface Handle
{
void doHandle(SelectionKey key) throws IOException;
} class AcceptHandler implements Handle
{ @Override
public void doHandle(SelectionKey key) throws IOException
{ final ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
final SocketChannel sc = ssc.accept();
final IOHandler handler = new IOHandler(key.selector(), sc);
System.out.println("server: connect success");
} } class IOHandler implements Handle
{
private final ByteBuffer readBuffer = ByteBuffer.allocate(1024); private OutputBuffer outputBuffer = new OutputBuffer(); private SocketChannel socketChannel = null; // private Selector selector = null; private SelectionKey key = null; public IOHandler(Selector selector, SocketChannel sc) throws IOException
{
// this.selector = selector;
this.socketChannel = sc;
socketChannel.configureBlocking(false);
key = socketChannel.register(selector, SelectionKey.OP_READ, this);
} /**
* 增加输出缓存
*
* @param writeData
* 要写出的数据
* @throws IOException
* @return 返回处理的字节数
*/
private int addWriteBuffer(ByteBuffer bytebuffer, int num) throws IOException
{
int prevPositon = bytebuffer.position();
outputBuffer.size += num; outputBuffer.writeBuffer.put(bytebuffer).flip();
int nowPosition = bytebuffer.position(); this.interestOps(0, SelectionKey.OP_WRITE); return nowPosition - prevPositon;
} /**
* 增加删除相应事件
*
* @param remove
* @param add
*/
private void interestOps(int remove, int add)
{
int cur = key.interestOps();
int ops = (cur & ~remove) | add;
if (cur != ops)
{
key.interestOps(ops);
key.selector().wakeup();
}
} /**
* ByteBuffer 转换 String
*
* @param buffer
* @return
*/
public static String getString(ByteBuffer buffer)
{
Charset charset = null;
CharsetDecoder decoder = null;
CharBuffer charBuffer = null;
try
{
charset = Charset.forName("UTF-8");
decoder = charset.newDecoder();
// charBuffer = decoder.decode(buffer);//用这个的话，只能输出来一次结果，第二次显示为空
charBuffer = decoder.decode(buffer.asReadOnlyBuffer());
return charBuffer.toString();
}
catch(Exception ex)
{
ex.printStackTrace();
return "";
}
} @Override
public void doHandle(SelectionKey key) throws IOException
{
if (key.isReadable())
{
System.out.print("server: meet read event ,before read position = " + readBuffer.position()); int num = socketChannel.read(readBuffer); // 关闭
if (num == -1)
{
System.out.println("close the channel ");
key.channel();
key.channel().close();
return;
} // 将position置为0
readBuffer.flip(); System.out.print(" reveive data " + getString(readBuffer)); int dealsize = addWriteBuffer(readBuffer, num); System.out.println(" write to writeBuffer size = " + dealsize + " nowPostion = " + readBuffer.position()); // 将处理过的数据清除
readBuffer.compact();
}
else if (key.isWritable())
{
System.out.print("meet write event"); long num = socketChannel.write(outputBuffer.writeBuffer);
outputBuffer.size -= num; System.out.print("deal size = " + num + "left buffer size = " + outputBuffer.size);
if (outputBuffer.size == 0)
{
System.out.println(" deal over,cancel write event");
interestOps(SelectionKey.OP_WRITE, 0);
}
// 清除已经处理过的数据
outputBuffer.writeBuffer.compact(); }
}
} class OutputBuffer
{
public int size; public final ByteBuffer writeBuffer = ByteBuffer.allocate(1024); }
ByteBuffer没有提供有用数据的相关方法，只能自己写一个OutputBuffer来辅助处理
之前OutputBuffer只是封装了一个ByteBuffer以及一个size变量用于标示可以数据量
下面对OutputBuffer进行了重构，将size变量的修改以及数据的写入和写出操作都封装到方法中，其中output（SocketChannel socketChannel）
方法利用回调的思想，将socketChannel对象传入，在OutputBuffer当中实现数据的write输出
package com.nio.echo; import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Iterator;
import java.util.Set; /**
* @author 作者 E-mail:
* @version 创建时间：2015-10-29 下午02:49:12 类说明
*/
public class NIOEchoServer
{
private static ServerSocketChannel ssc = null; private static Selector selector = null; private static final int PORT = 8080; public static void startServer() throws IOException
{
ssc = ServerSocketChannel.open();
selector = Selector.open();
ssc.configureBlocking(false); // nio 对socket 和serverSocket进行了怎样封装
final ServerSocket serverSocket = ssc.socket(); serverSocket.bind(new InetSocketAddress(PORT));
serverSocket.setReuseAddress(true); final AcceptHandler acceptHandler = new AcceptHandler();
ssc.register(selector, SelectionKey.OP_ACCEPT, acceptHandler);
while (true)
{
int n = selector.select();
if (n == 0)
continue; final Set<SelectionKey> readyKeys = selector.selectedKeys();
final Iterator<SelectionKey> it = readyKeys.iterator();
while (it.hasNext())
{
final SelectionKey key = it.next();
final Handle handler = (Handle) key.attachment();
handler.doHandle(key);
it.remove();
}
}
} public static void main(String[] args) throws IOException
{
NIOEchoServer.startServer();
}
} interface Handle
{
void doHandle(SelectionKey key) throws IOException;
} class AcceptHandler implements Handle
{ @Override
public void doHandle(SelectionKey key) throws IOException
{ final ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
final SocketChannel sc = ssc.accept();
final IOHandler handler = new IOHandler(key.selector(), sc);
System.out.println("server: connect success");
} } class IOHandler implements Handle
{
private final ByteBuffer readBuffer = ByteBuffer.allocate(1024); private OutputBuffer outputBuffer = new OutputBuffer(); private SocketChannel socketChannel = null; // private Selector selector = null; private SelectionKey key = null; public IOHandler(Selector selector, SocketChannel sc) throws IOException
{
// this.selector = selector;
this.socketChannel = sc;
socketChannel.configureBlocking(false);
key = socketChannel.register(selector, SelectionKey.OP_READ, this);
} /**
* 增加输出缓存
*
* @param writeData
* 要写出的数据
* @throws IOException
* @return 返回处理的字节数
*/
private int addWriteBuffer(ByteBuffer bytebuffer, int num) throws IOException
{
int prevPositon = bytebuffer.position(); outputBuffer.put(bytebuffer, num); int nowPosition = bytebuffer.position(); this.interestOps(0, SelectionKey.OP_WRITE); return nowPosition - prevPositon;
} /**
* 增加删除相应事件
*
* @param remove
* @param add
*/
private void interestOps(int remove, int add)
{
int cur = key.interestOps();
int ops = (cur & ~remove) | add;
if (cur != ops)
{
key.interestOps(ops);
key.selector().wakeup();
}
} /**
* ByteBuffer 转换 String
*
* @param buffer
* @return
*/
public static String getString(ByteBuffer buffer)
{
Charset charset = null;
CharsetDecoder decoder = null;
CharBuffer charBuffer = null;
try
{
charset = Charset.forName("UTF-8");
decoder = charset.newDecoder();
// charBuffer = decoder.decode(buffer);//用这个的话，只能输出来一次结果，第二次显示为空
charBuffer = decoder.decode(buffer.asReadOnlyBuffer());
return charBuffer.toString();
}
catch(Exception ex)
{
ex.printStackTrace();
return "";
}
} @Override
public void doHandle(SelectionKey key) throws IOException
{
if (key.isReadable())
{
System.out.print("server: meet read event ,before read position = " + readBuffer.position()); int num = socketChannel.read(readBuffer); // 关闭
if (num == -1)
{
System.out.println("close the channel ");
key.channel();
key.channel().close();
return;
} // 将position置为0
readBuffer.flip(); System.out.println(" reveive data " + getString(readBuffer)); int dealsize = addWriteBuffer(readBuffer, num); // 将处理过的数据清除
readBuffer.compact();
}
else if (key.isWritable())
{
System.out.print("meet write event"); // 写数据
outputBuffer.output(socketChannel); if (outputBuffer.size() == 0)
{
System.out.println(" deal over,cancel write event");
interestOps(SelectionKey.OP_WRITE, 0);
} }
}
} class OutputBuffer
{
private int size; private final ByteBuffer writeBuffer = ByteBuffer.allocate(1024); public void output(SocketChannel socketChannel) throws IOException
{
int num = socketChannel.write(writeBuffer);
writeBuffer.compact();
size -= num;
} public void put(ByteBuffer b, int num)
{
writeBuffer.put(b).flip();
this.size += num;
} public int size()
{
return this.size;
}
}
事实上在NIO网络编程中，写出数据的操作需要加入缓存才能保证效率，目的是为了写操作发生的时候不影响业务继续send消息，首先将send消息发送过来的数据缓存到A中，在写事件发生的时候将A中数据写出(此时仅短暂锁住A，将A中引用拿出，重新赋值新引用给A)，这样写事件的处理过程和业务消息的send就可以高并发的进行。