转载请注明出处
http://blog.csdn.net/sinat_25295611/article/details/77871116
By Kay 2017.09.06
关于Java NIO 基础，推荐 IBM developerWorks上的一篇文章，写的非常的好，作者是Greg Travis 。
NIO入门
https://www.ibm.com/developerworks/cn/education/java/j-nio/j-nio.html
NIO Socket Demo
这里贴一个我练习的一个NIO例子，该例子是在阅读学习《Netty权威指南 》这本书参考写的，为的是理解NIO的思想。
关键点我都写在注释中，特别要注意 I/O多路复用模型，把大量的I/O请求连接复用到一个Selector线程中去处理。
以下为例子：
场景：客户端向服务端发送一条请求当前时间的指令，服务端收到指令后返回当前时间给客户端输出。
客户端：TimeClient
服务端：TimeServer
TimeServer
/**
* Created by kay on 2017/9/8.
*/
public class TimeServer {
public static void main(String[] args) {
int port=8888;
MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
new Thread(timeServer,"多路复用TimeServer启动").start();
}
}
MultiplexerTimeServer 多路复用类，也就是服务端的处理线程
package com.kay.concurrent.nio.timesocket;
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.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;
/**
* Created by kay on 2017/9/7.
* 多路复用类
* 处理多个客户端的并发请求
* selector 多路复用器
*/
public class MultiplexerTimeServer implements Runnable{
private Selector selector;
private ServerSocketChannel servChannel;
private volatile boolean stop;
/**
* 初始化 绑定注册监听
* @param port
*/
public MultiplexerTimeServer(int port) {
try {
selector=Selector.open();
servChannel = ServerSocketChannel.open();
servChannel.configureBlocking(false); //设置非阻塞模式
servChannel.bind(new InetSocketAddress(port)); //绑定端口
servChannel.register(selector, SelectionKey.OP_ACCEPT); //监听准备连接
System.out.println("TimeServer 正在监听端口："+port);
} catch (IOException e) {
e.printStackTrace();
//初始化失败则退出，例如端口占用等
System.exit(1);
}
}
@Override
public void run() {
while (!stop) {
try {
//每隔一秒 轮询一次
selector.select(1000);
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectionKeys.iterator();
while (it.hasNext()) {
SelectionKey key=it.next();
it.remove();
try {
//处理准备好的事件
handleInput(key);
} catch (Exception e) {
if(key!=null){
key.cancel();
if (key.channel() != null) {
key.channel().close();
}
}
}
}
} catch (Throwable t) {
t.printStackTrace();
}
}
//关闭多路复用器，绑定在上面的channel也会被自动关闭
if (selector != null) {
try {
selector.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 统一事件处理，根据key 的类型作出相应处理
* 1.对方请求连接->accept->注册监听对方发的消息
* 2.对方发送消息->读取消息->做出响应
* @param key
* @throws IOException
*/
private void handleInput(SelectionKey key) throws IOException{
//判断是否可用
if (key.isValid()) {
//判断是否是accept事件
if (key.isAcceptable()) {
//拿到这个key上面绑定的Channel，然后获取对面来的SocketChannel
//将这个channel注册 监听它的读事件（因为它已经连接了，所以就等着它发消息了）
ServerSocketChannel ssc= (ServerSocketChannel) key.channel();
SocketChannel sc=ssc.accept();
sc.configureBlocking(false);
System.out.println("--新请求接入,开始监听它发来的消息...");
sc.register(selector, SelectionKey.OP_READ);
}
//判断对方是否放消息来了，是就读取消息/作出响应
if (key.isReadable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = socketChannel.read(readBuffer);
if (readBytes > 0) {
readBuffer.flip();
byte[] bytes = new byte[readBuffer.remaining()];
readBuffer.get(bytes);
String body = new String(bytes, "UTF-8");
System.out.println("Time Server 接收到消息："+body);
String currentTime="";
if ("QUERY_TIME".equals(body)) {
currentTime = "现在时间是：" + new Date(System.currentTimeMillis()).toString();
} else {
currentTime="指令错误！";
}
//作出响应
doWrite(socketChannel,currentTime);
} else if (readBytes < 0) {
key.cancel();
socketChannel.close();
}else {
;//读到0字节忽略
}
}
}
}
//响应消息
private void doWrite(SocketChannel socketChannel, String response) throws IOException {
if (response != null && response.trim().length() > 0) {
byte[] bytes=response.getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
socketChannel.write(writeBuffer);
System.out.println("--已发送响应");
}
}
}
TimeClient 客户端
/**
* Created by kay on 2017/9/8.
*/
public class TimeCient {
public static void main(String[] args) {
int port=8888;
TimeClientHandle clientHandle=new TimeClientHandle("127.0.0.1",port);
new Thread(clientHandle,"TimeClient").start();
}
}
TimeClientHandle 客户端处理线程
package com.kay.concurrent.nio.timesocket;
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;
import java.util.Set;
/**
* Created by kay on 2017/9/8.
*/
public class TimeClientHandle implements Runnable {
private String host;
private int port;
private SocketChannel socketChannel;
private Selector selector;
private volatile boolean stop=false;
public TimeClientHandle(String host,int port) {
this.host=(host==null )? "127.0.0.1":host;
this.port=port;
try {
selector=Selector.open();
socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
@Override
public void run() {
try {
doConnect();
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
while (!stop) {
try {
selector.select(1000);
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectionKeys.iterator();
SelectionKey key=null;
while (it.hasNext()) {
key=it.next();
it.remove();
try {
handleInput(key);
} catch (Exception e) {
if (key != null) {
key.cancel();
if (key.channel() != null) {
key.channel().close();
}
}
}
}
} catch (Exception e) {
e.printStackTrace();
System.exit(1);
}
}
if (selector != null) {
try {
selector.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
private void handleInput(SelectionKey key) throws IOException {
SocketChannel sc= (SocketChannel) key.channel();
if (key.isConnectable()) {
if (sc.finishConnect()) {
sc.register(selector, SelectionKey.OP_READ);
doWrite(sc);
}else
System.exit(1);
}
if (key.isReadable()) {
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(readBuffer);
if (readBytes > 0) {
readBuffer.flip();
byte[] bytes = new byte[readBuffer.remaining()];
readBuffer.get(bytes);
String body = new String((bytes), "UTF-8");
System.out.println("--接收到消息:" + body);
this.stop=true;
} else if (readBytes < 0) {
//对面关掉了链接
key.cancel();
sc.close();
}else
; //没有读到东西
}
}
private void doConnect() throws IOException {
//客户端去连接服务器，如果返回false，则说明发送了syn，但服务器没有响应ack，三次握手没有完成
if (socketChannel.connect(new InetSocketAddress(host, port))) {
socketChannel.register(selector, SelectionKey.OP_READ);
doWrite(socketChannel);
} else {
//如果没有直连成功，则新建连接
socketChannel.register(selector, SelectionKey.OP_CONNECT);
}
}
private void doWrite(SocketChannel sc) throws IOException {
byte[] req="QUERY_TIME".getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
writeBuffer.put(req);
writeBuffer.flip();
sc.write(writeBuffer);
if (!writeBuffer.hasRemaining()) {
System.out.println("指令发送成功！");
}
}
}
