Java NIO是一种基于通道和缓冲区的I/O方式，已经被广泛的应用，成为解决高并发与大量连接和I/O处理问题的有效方式。
Java NIO相关组件
Java NIO主要有三个核心部分组成，分别是：Channel(通道)，Buffer(缓冲区), Selector（选择器）
- Channel
Channel是所有访问IO设备的统称。类型与IO中的Stream，而通道是双向的，既可以读又可以写，但是Stream是单项的。常用的通道有：SocketChannel和ServerSocketChannel（对应TCP的客户端和服务器端）、FileChannel（对应文件IO）、DatagramChannel（对应UDP）等
- Buffer
所有数据的读写都要经过Buffer，Buffer直接和Channel打交道，是一个存储数据的容器。通过调用Channel.write方法将数据写入Buffer，Channel.read方法将数据从Buffer中读取出来。常用的Buffer有：ByteBuffer、LongBuffer、IntBuffer、StringCharBuffer等
- Selector
Selector用来监听多个Channel的事件（比如：Read、Write、Connect和Accept等），通过单个线程轮询的方式实现了对多个Channel的监听。
Java IO与NIO的区别
NIO是一种叫非阻塞IO（Non-blocking I/O），基于I/O多路复用来实现的（可参考：I/O模型、select、poll和epoll之间的区别）。NIO与之前传统的I/O模型有很大的不同，具体表现在以下几个方面：
- 面向流与面向缓冲
Java IO和NIO之间一个最大的区别是，IO是面向流的，NIO是面向缓冲区的。Java IO每次从数据流中读一个或多个字节，直至读取所有字节，数据流是一次性的，读取完以后，不能前后移动流中的数据。Java NIO是将数据读取到缓冲区，可以通过position来回移动访问缓冲区中的数据。
- 阻塞与非阻塞IO
Java IO中调用read和write方法的线程会被阻塞的，直到数据全部读入或者全部写入完为止。而在Java NIO中，如果需要读写数据只用和缓冲区打交道，将数据从缓冲区读取或者写入缓冲区以后，线程可以继续做其他事情，不会被block住。
- 选择器（Selector）
Selector是基于I/O多路复用的机制实现的，将多个Channel注册到一个Selector上，Selector通过轮询监听所有注册的通道上是否有SelectionKey发生，如果发生了，然后将SelectionKey分派给其他线程处理。
Java NIO的应用
通过Java NIO技术简单实现了一个服务端与客户端通信的case，具体功能如下：
- 服务端可以向客户端广播消息
- 服务端将一个客户端的消息转发给其他客户端
- 客户端向服务端发送消息
- 客户端接收服务端的消息
服务端代码如下：
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Server {
public static void main(String[] args) throws IOException {
new Server().start(); // 启动服务端程序
}
public Server() throws IOException {
this.init(); // 初始化服务端数据
}
/**
* 服务端端口
*/
private int port = 9999;
/**
* 服务端的Selector用来监听Channel的事件.
*/
private Selector selector;
/**
* 字符数据编码
*/
private Charset charset = Charset.forName("UTF-8");
/**
* 读缓存，分配1024Byte的空间
*/
private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
/**
* 写缓存，分配1024Byte的空间
*/
private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
/**
* 存储所有客户端的Channel，转发的时候使用
*/
private Map<String, Channel> clientSocketChannels = new HashMap<>();
/**
* 定义了一个线程池，服务端用来发送数据给客户端
*/
private static ExecutorService executorService = Executors.newFixedThreadPool(1, runnable -> {
Thread thread = new Thread(runnable);
thread.setDaemon(true);
thread.setName("server-sender");
return thread;
});
/**
* 初始化Channel.
*/
private void init() throws IOException {
// 声明一个服务端的ServerSocketChannel
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 将服务端的ServerSocketChannel设置成非阻塞模式
serverSocketChannel.configureBlocking(false);
// 设置服务端的socket
ServerSocket serverSocket = serverSocketChannel.socket();
serverSocket.bind(new InetSocketAddress(this.port));
// 声明一个Selector，用来监听服务端的所有Channel
this.selector = Selector.open();
// 在ServerSocketChannel上注册Accept事件，用来接收客户端的连接
serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
System.out.println("Server is started, the port is " + this.port);
}
/**
* 处理服务端监听到的事件
*/
private void work(SelectionKey selectionKey) throws IOException {
// 客户端有Socket连接请求
if (selectionKey.isAcceptable()) {
// 从SelectionKey中获取服务端的ServerSocketChannel，SelectionKey中包含了服务端与客户端的所有信息
ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
// 服务端打开一个新的SocketChannel用来与客户端的SocketChannel进行通信，服务端同时会随机分配一个端口
SocketChannel socketChannel = serverSocketChannel.accept();
// 将SocketChannel设置成非阻塞模式
socketChannel.configureBlocking(false);
// 将SocketChannel中的Read事件注册到Selector上
socketChannel.register(this.selector, SelectionKey.OP_READ);
// 存储服务端为客户端创建的SocketChannel，为后面的转发消息服务
this.clientSocketChannels.put(this.getClientName(socketChannel), socketChannel);
// 通过System.in IO流来创建Scanner
Scanner scanner = new Scanner(System.in);
// 收集服务端控制台输入的数据，通过线程池将数据广播给所有客户端SocketChannel
this.executorService.submit(() -> {
while (true) {
// 该方法会被block住，一直等到服务端控制台有数据输入完为止
String sendText = scanner.nextLine();
// 将服务端的数据广播给所有客户端
transferToOthers(sendText, null);
}
});
// 服务端监听到有数据可以读取，主要是来源于客户端发送的数据
} else if (selectionKey.isReadable()) {
// 获取服务端的SocketChannel，然后与客户端进行通信
// 需要注意的是：当前获取的SocketChannel与ServerSocketChannel是不同的，
// 这个SocketChannel是通过调用ServerSocketChannel.accept方法创建的（存储在clientSocketChannels集合中）
SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
// 清空当前的用来存储读数据的buffer
readBuffer.clear();
// 将数据从SocketChannel读入buffer
int bytes = socketChannel.read(readBuffer);
if (bytes > 0) {
// 使得buffer中的数据可读
readBuffer.flip();
// 读取buffer中的数据
String text = String.valueOf(this.charset.decode(readBuffer));
System.out.println(this.getClientName(socketChannel) + ": " + text);
// 将客户端发送过来的数据转发给其他客户端
this.transferToOthers(text, socketChannel);
}
}
}
/**
* 将数据发送给其他客户端
*/
private void transferToOthers(String text, final SocketChannel socketChannel) {
this.clientSocketChannels.forEach((channelName, channel) -> {
// 获取之前存储的与服务端建立连接的客户端
SocketChannel otherSocketChannel = (SocketChannel) channel;
if (!otherSocketChannel.equals(socketChannel)) {
// 清空写缓存
this.writeBuffer.clear();
// 将数据写入缓存
this.writeBuffer.put(this.charset.encode(this.getClientName(socketChannel) + ": " + text));
// 使得缓存中的数据变得可用
this.writeBuffer.flip();
try {
// 将buffer中的数据写入到其它客户端
otherSocketChannel.write(this.writeBuffer);
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
/**
* 通过SocketChannel生成客户端的名字，用来标识
*/
private String getClientName(SocketChannel socketChannel) {
if (socketChannel == null)
return "[server]";
Socket socket = socketChannel.socket();
return "[" + socket.getInetAddress().toString().substring(1) + ":" + socket.getPort() + "]";
}
/**
* 启动服务端程序
*/
public void start() {
// 无限循环来轮询所有注册的Channel
while (true) {
try {
// 选择已经准备好的Channel，该方法是会block住的，直到有事件到达
this.selector.select();
// 获取所有监听到的事件
Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
while (iterator.hasNext()) {
// 找到事件SelectionKey，里面包含了事件相关的所有数据
SelectionKey selectionKey = iterator.next();
// 如果事件是有效的
if (selectionKey.isValid()) {
// 处理事件
this.work(selectionKey);
}
// 删除已经处理过的事件
iterator.remove();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
客户端代码如下：
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
public class Client {
public static void main(String[] args) throws IOException {
new Client().start(); // 客户端程序执行入口
}
/**
* 注册监听的服务的端口，并初始化
*/
public Client() throws IOException {
this.serverSocketAddress = new InetSocketAddress("127.0.0.1", 9999);
this.init();
}
/**
* 服务的Socket地址
*/
private SocketAddress serverSocketAddress;
/**
* 客户端Selector
*/
private Selector selector;
/**
* 字符编码
*/
private Charset charset = Charset.forName("UTF-8");
/**
* 读缓冲区
*/
private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
/**
* 写缓冲区
*/
private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
/**
* 线程池执行客户端发送数据
*/
private static ExecutorService executorService = Executors.newFixedThreadPool(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable runnable) {
Thread thread = new Thread(runnable);
thread.setDaemon(true);
thread.setName("client-sender");
return thread;
}
});
/**
* 初始化客户端信息
*/
private void init() throws IOException {
// 声明一个客户端SocketChannel
SocketChannel socketChannel = SocketChannel.open();
// 设置成非阻塞模式
socketChannel.configureBlocking(false);
// 声明一个Selector
this.selector = Selector.open();
// 将客户端的SocketChannel的连接事件注册到selector上
socketChannel.register(this.selector, SelectionKey.OP_CONNECT);
// 连接服务端
socketChannel.connect(this.serverSocketAddress);
}
/**
* 处理客户端数据
*/
private void work(SelectionKey selectionKey) {
try {
// 与服务端建立连接
if (selectionKey.isConnectable()) {
// 从SelectionKey中获取客户端的ServerSocketChannel
SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
// 判断连接是否完成
if (socketChannel.isConnectionPending()) {
// 完成连接
socketChannel.finishConnect();
System.out.println("The connection is successful！");
// 通过System.in IO流来创建Scanner
Scanner scanner = new Scanner(System.in);
// 使用线程池来完成对客户端的控制台数据输入的监听
executorService.submit((Runnable) () -> {
while (true) {
try {
// 清空写缓冲区
writeBuffer.clear();
// 该方法会被block住，一直等到客户端控制台有数据输入完为止
String sendText = scanner.nextLine();
// 将数据写入写缓冲区
writeBuffer.put(charset.encode(sendText));
// 使得写缓冲区中的数据可读
writeBuffer.flip();
// 将数据通过SocketChannel发送到服务端
socketChannel.write(writeBuffer);
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
// 注册可读事件，应该当前的SocketChannel与服务端建立连接以后，不需要再监听创建连接的事件
// 为了复用SocketChannel，将SocketChannel的Read事件注册到Selector
socketChannel.register(selector, SelectionKey.OP_READ);
}
// 可读事件，有从服务器端发送过来的信息，读取输出到控制台上
else if (selectionKey.isReadable()) {
// 获取与服务端通信的客户端SocketChannel
SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
// 清空读缓冲区
this.readBuffer.clear();
// 将数据读取到读缓冲区，并将数据输出到客户端控制台
int count = socketChannel.read(this.readBuffer);
if (count > 0) {
String text = new String(this.readBuffer.array(), 0, count);
System.out.println(text);
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 启动客户端程序
*/
public void start() throws IOException {
// 无限循环，轮询所有监听的SocketChannel
while (true) {
// 选择已经准备好的Channel，该方法是会block住的，直到有事件到达
int events = this.selector.select();
if (events > 0) {
// 找到事件SelectionKey，里面包含了事件相关的所有数据
Set<SelectionKey> selectionKeys = selector.selectedKeys();
// 处理事件
selectionKeys.forEach(selectionKey -> this.work(selectionKey));
// 清空已处理的事件
selectionKeys.clear();
}
}
}
}
总结
- 服务端的
ServerSocketChannel是用来监听客户端的连接请求，只有1个且端口固定，主要监听accept事件 - 服务端的
SocketChannel是用来和客户端建立数据读写操作通信，数量与客户端的连接数量一致，每个都分配一个随机的端口，主要监听read事件 - 每个客户端有一个
SocketChannel，用来和服务端进行通信，主要监听connect事件和read事件，connect事件只会在第一连接时发生，read事件是在每次接收服务端数据时发生 - 服务端和客户端各有一个
Selector，用来监听所有的SocketChannel或者ServerSocketChannel中注册的事件，在没有事件发生的时候，Selector.select()会被block住 - 在定义缓冲区的时候要注意缓冲区的大小，如果太小会报
BufferOverflowException