前言
这周学习尼恩编著的《Netty、Redis、ZooKeeper高并发实战》, 这本书写的很不错，通过十几个例子带领大家去体会高并发如何实现， 这周我看了最基础的JavaNOI部分，读书的时候好像明白了作者写的内容，但是又体会不深，非得自己动手写一些书上得例子，有时候还要改动下例子，才能体会深刻，得出自己得结论。下面我们进入例子演示。
Blocking IO
首先我们会来一个Blocking IO的例子，也就是同步阻塞方式，在Java中，默认创建的socket都是阻塞的。
先来看Server端代码， 写一个继承ServerSocket的类
@Slf4j
public class BlockReceiveServer extends ServerSocket {
public BlockReceiveServer() throws Exception
{
super(SERVER_PORT);
}
public void startServer() throws Exception
{
while (true)
{
// server尝试接收其他Socket的连接请求，server的accept方法是阻塞式的
log.debug("server listen at:" + SERVER_PORT);
Socket socket = this.accept();
/**
* 我们的服务端处理客户端的连接请求是同步进行的， 每次接收到来自客户端的连接请求后，
* 都要先跟当前的客户端通信完之后才能再处理下一个连接请求。 这在并发比较多的情况下会严重影响程序的性能，
* 为此，我们可以把它改为如下这种异步处理与客户端通信的方式
*/
// 每接收到一个Socket就建立一个新的线程来处理它
new Thread(new Task(socket)).start();
}
}
this.accept()是阻塞的，也就是说没有请求来的时候，它是停在这里的，有请求来了后，我们起一个新的线程来出来它。 这里定义了一个Task来出来客户端请求。
class Task implements Runnable
{
private Socket socket;
private DataInputStream dis;
public Task(Socket socket)
{
this.socket = socket;
}
@Override
public void run()
{
try
{
dis = new DataInputStream(socket.getInputStream());
log.debug("start receive");
for (int i = 0;i < 1000 ;i++)
{
dis.readUTF();
}
log.debug("finish receive" );
} catch (Exception e)
{
e.printStackTrace();
} finally
{
IOUtil.closeQuietly(dis);
IOUtil.closeQuietly(socket);
}
}
}
书中的例子是传送文件，我这里把它改为接收客户端传过来的文本，就是客户端写一千次，服务端读一千次，这个也是写死的，因为阻塞，没有读完，你是不能进入下一个环节的，不然会出错。
客户端代码
@Slf4j
public class BlockSendClient extends Socket
{
private Socket client;
private DataOutputStream outputStream;
/**
* 构造函数<br/>
* 与服务器建立连接
*
* @throws Exception
*/
public BlockSendClient() throws Exception
{
super(SOCKET_SERVER_IP
, SERVER_PORT);
this.client = this;
}
/**
* 向服务端传输文件
*
* @throws Exception
*/
public void sendFile() throws Exception
{
try
{
outputStream = new DataOutputStream(client.getOutputStream());
for (int i = 0;i < 1000 ;i++)
{
outputStream.writeUTF(String.valueOf(i));
outputStream.flush();
Thread.sleep(1000);
}
log.debug("======== file transfer success ========");
} catch (Exception e)
{
e.printStackTrace();
} finally
{
IOUtil.closeQuietly(outputStream);
IOUtil.closeQuietly(client);
}
}
客户端就是继承了socket,然后往socket里面发送1000次数字。
为了体现效果，我把客户端也让它多线程的去和服务端连接。 这里启动50个线程去和服务端连接，如果启动100个，会报“java.net.ConnectException: Connection refused: connect”， 但是你多启动几个客户端去连接，是不会有问题的。
public static void main(String[] args)
{
for(int i =0;i<50;i++){
new Thread(new Task()).start();
}
}
class Task implements Runnable
{
@Override
public void run() {
try
{
BlockSendClient client = new BlockSendClient(); // 启动客户端连接
client.sendFile();
} catch (Exception e)
{
e.printStackTrace();
}
}
}
我们可以看到服务端的输出
我们打开多少客户端，那么服务端就必须要打开50倍的线程数来连接。 而线程能开的数量是有限制的，一个Java线程大约占1M的空间。 对于高并发，这种方式肯定是不可取的。
Java NIO
我们来看下如何用Java NOI 来写一个和上面一样的功能。先来看下服务端代码
@Slf4j
public class NioDiscardServer {
public static void startServer() throws IOException {
// 1、创建一个 Selector选择器
Selector selector = Selector.open();
// 2、获取通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
ServerSocket serverSocket = serverSocketChannel.socket();
// 3.设置为非阻塞
serverSocketChannel.configureBlocking(false);
// 4、绑定监听端口
serverSocket.bind(new InetSocketAddress(SERVER_PORT));
log.info("server start success");
// 5、将通道注册到选择器上,并注册的IO事件为：“接收新连接”
SelectionKey sk = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
// sk.interestOps(SelectionKey.OP_ACCEPT) ;
// 6、轮询感兴趣的I/O就绪事件（选择键集合）
while (selector.select() > 0) {
// 7、获取选择键集合
Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
while (selectedKeys.hasNext()) {
// 8、获取单个的选择键，并处理
SelectionKey selectedKey = selectedKeys.next();
// 9、判断key是具体的什么事件
if (selectedKey.isAcceptable()) {
log.info("new connection coming" + selectedKey.channel());
ServerSocketChannel server = (ServerSocketChannel) selectedKey.channel();
SocketChannel socketChannel = server.accept();
// 10、若选择键的IO事件是“连接就绪”事件,就获取客户端连接
// 11、切换为非阻塞模式
socketChannel.configureBlocking(false);
socketChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
// 12、将该通道注册到selector选择器上
SelectionKey channleSk= socketChannel.register(selector,
SelectionKey.OP_READ | SelectionKey.OP_WRITE | SelectionKey.OP_CONNECT);
}
if (selectedKey.isWritable()) {
//log.info("write ready:" + selectedKey.channel());
}
if (selectedKey.isConnectable()) {
log.info("client connect success:" + selectedKey.channel());
}
if (selectedKey.isReadable()) {
log.info("read ready:" + selectedKey.channel());
// 13、若选择键的IO事件是“可读”事件,读取数据
SocketChannel socketChannel = (SocketChannel) selectedKey.channel();
// 14、读取数据
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
int length = 0;
while ((length = socketChannel.read(byteBuffer)) > 0) {
byteBuffer.flip();
log.info(new String(byteBuffer.array(), 0, length));
byteBuffer.clear();
}
if(length == -1)
{
socketChannel.close();
}
}
// 15、移除选择键
selectedKeys.remove();
}
}
// 7、关闭连接
serverSocketChannel.close();
}
public static void main(String[] args) throws IOException {
startServer();
}
}
客户端代码如下
@Slf4j
public class NioDiscardClient {
public void startClient(int clientNum) throws IOException {
InetSocketAddress address =
new InetSocketAddress(SOCKET_SERVER_IP,
SERVER_PORT);
// 1、获取通道（channel）
SocketChannel socketChannel = SocketChannel.open(address);
// 2、切换成非阻塞模式
socketChannel.configureBlocking(false);
//不断的自旋、等待连接完成，或者做一些其他的事情
while (!socketChannel.finishConnect()) {
}
log.info("client connect success");
for(int i = 0;i<1000;i++) {
// 3、分配指定大小的缓冲区
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
byteBuffer.put((clientNum + "hello" + i).getBytes());
byteBuffer.flip();
socketChannel.write(byteBuffer);
ThreadUtil.sleepSeconds(1);
}
log.info("client write success");
socketChannel.shutdownOutput();
socketChannel.close();
}
public static void main(String[] args) throws IOException {
for(int i =0;i<50;i++){
new Thread(new NoiTask(i)).start();
}
}
}
class NoiTask implements Runnable
{
private int clientNum;
public NoiTask(int clientNum)
{
this.clientNum = clientNum;
}
@Override
public void run() {
try
{
NioDiscardClient client = new NioDiscardClient(); // 启动客户端连接
client.startClient(clientNum); // 传输文件
} catch (Exception e)
{
e.printStackTrace();
}
}
}
代码这里不解释了。 这里服务端和客户端不需要约定多少内容要发送，
服务端有段代码需要注意
while ((length = socketChannel.read(byteBuffer)) > 0)，这段很容易就退出循环，因为read是异步的，很容易就返回0， 让循环中断， 开始的时候socketChannel.close()没有加上条件，服务端打印出几次后，就close了，客户端就再写不进去了。报“java.io.IOException: An established connection was aborted by the software in your host machine”，因为服务端关闭了。
服务端的输出我们可以看出，它只启动了一个main线程，不论多少个客户端连接过来。
我再用CMD启动多个进行客户端进行测试的时候，发现它会启用300多M的内存，后面多加几个客户端，它也不会增长，保持再这个数字上面。 前面的例子中，内存使用的要小，但是增加客户端后，内存使用量会增加，但是也不是很明显。 这里面为什么会用这么多内存，我不是太明白，也许代码有什么缺陷在这里，暂时不细究了。
总结
我在使用NIO的时候，发现它在单机测试性能的时候不比传统的阻塞式的优秀，甚至还慢一些。它的应用场景是在量大的时候，高并发的时候。具体NIO的原理我也解释不清楚，只能从应用的层面来用例子实践一下，后面看能不能再深入下它的原理，书中讲述的很好，但是我只能说大致理解，谈不上深入。
