Java NIO —— TCP套接字（ServerSocketChannel & SocketChannel）

package com.demo.test;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SocketChannelDemo {

private static final int PORT = 9999;
private static final int ACCEPT_INTERVAL = 300;
private static final int BUFF_SIZE = 255;
private static final int WAIT_CONNECT_INTERVAL = 300;

private static void openServerSocketChannel(boolean isBlocking) {
try {
// The new channel's socket is initially unbound; it must be bound to a specific address via one of its socket's bind methods before connections can be accepted.
final ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(PORT));
serverSocketChannel.configureBlocking(isBlocking);
while (true) {
final SocketChannel socketChannel = serverSocketChannel.accept();
// 在非阻塞模式下，accept() 方法会立刻返回，如果还没有新进来的连接,返回的将是null
if (!isBlocking && socketChannel == null) {
System.out.println("serverSocketChannel.accept() = null");
try {
Thread.sleep(ACCEPT_INTERVAL);
} catch (InterruptedException e) {
e.printStackTrace();
}
continue;
}
System.out.println("serverSocketChannel.accept() != null");
new Thread(new Runnable() {

@Override
public void run() {
// do something with socketChannel...
ByteBuffer readBuffer = ByteBuffer.allocate(BUFF_SIZE);
byte[] buffer = new byte[BUFF_SIZE];
int len = -1;
do {
try {
len = socketChannel.read(readBuffer);
} catch (IOException e) {
e.printStackTrace();
try {
socketChannel.close();
} catch (IOException e1) {
e1.printStackTrace();
}
break;
}
if (len != -1) {
//切换读写模式，socketChannel.read是向ByteBuffer里面写，而get是将数据从ByteBuffer里读出来
readBuffer.flip();
//get也会改变position
readBuffer.get(buffer, 0, len);
//rewind会将position的位置设置为0
readBuffer.rewind();
try {
socketChannel.write(readBuffer);
} catch (IOException e) {
e.printStackTrace();
}
//将当前指针移到数组首位，相当于清空数据
readBuffer.clear();
System.out.println("service recv : len=" + len + ", data=" + new String(buffer, 0, len));
}
} while (len != -1);
}
}).start();
break;
}
try {
Thread.sleep(4000);
} catch (InterruptedException e) {
e.printStackTrace();
}
serverSocketChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
}

private static void openSocketChannel(boolean isBlocking) {
try {
final SocketChannel socketChannel = SocketChannel.open();
// true if a connection was established, false if this channel is in non-blocking mode and the connection operation is in progress
boolean isEstablished = socketChannel.connect(new InetSocketAddress("127.0.0.1", PORT));
System.out.println("socketChannel.connect() = " + isEstablished);
if (!isBlocking && !isEstablished) {
while (!socketChannel.finishConnect()) {
// wait, or do something else...
System.out.println("socketChannel.finishConnect() = false");
try {
Thread.sleep(WAIT_CONNECT_INTERVAL);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("socketChannel.finishConnect() = true");
}
new Thread(new Runnable() {

@Override
public void run() {
int len = -1;
do {
ByteBuffer readBuffer = ByteBuffer.allocate(BUFF_SIZE);
byte[] buffer = new byte[BUFF_SIZE];
// The number of bytes read, possibly zero, or -1 if the channel has reached end-of-stream
try {
len = socketChannel.read(readBuffer);
if (len != -1) {
readBuffer.flip();
readBuffer.get(buffer, 0, len);
readBuffer.compact();
System.out.println("client recv : len=" + len + ", data=" + new String(buffer, 0, len));
}
} catch (IOException e) {
e.printStackTrace();
break;
}
} while (len != -1);
}
}).start();
ByteBuffer writeBuffer = ByteBuffer.allocate(BUFF_SIZE);
for (int i = 0; i < 5; i++) {
String str = "(" + i + ")" + new Date().toString();
byte[] buffer = str.getBytes();
//将数组写入到ByteBuffer中
writeBuffer.put(buffer);
// 这里要交换读写模式
writeBuffer.flip();
//将数据从ByteBuffer中读出，写入到流中
socketChannel.write(writeBuffer);
// 压缩数据，即将数据向前移动已使用的长度， 如果没有这句会报 java.nio.BufferOverflowException
writeBuffer.compact();
System.out.println("client send : len=" + buffer.length + ", data=" + str);
}
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
socketChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
}

public static void main(String[] args) {
int nThreads = 2;
ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
final boolean isBlocking = false;
executorService.execute(new Runnable() {

@Override
public void run() {
openServerSocketChannel(isBlocking);
}
});
executorService.execute(new Runnable() {

@Override
public void run() {
openSocketChannel(isBlocking);
}
});
}
}

serverSocketChannel.accept() != null
socketChannel.connect() = true
client send : len=31, data=(0)Mon Dec 05 09:25:48 CST 2016
client send : len=31, data=(1)Mon Dec 05 09:25:48 CST 2016
client send : len=31, data=(2)Mon Dec 05 09:25:48 CST 2016
client send : len=31, data=(3)Mon Dec 05 09:25:48 CST 2016
client send : len=31, data=(4)Mon Dec 05 09:25:48 CST 2016
service recv : len=31, data=(0)Mon Dec 05 09:25:48 CST 2016
service recv : len=124, data=(1)Mon Dec 05 09:25:48 CST 2016(2)Mon Dec 05 09:25:48 CST 2016(3)Mon Dec 05 09:25:48 CST 2016(4)Mon Dec 05 09:25:48 CST 2016
client recv : len=31, data=(0)Mon Dec 05 09:25:48 CST 2016
client recv : len=124, data=(1)Mon Dec 05 09:25:48 CST 2016(2)Mon Dec 05 09:25:48 CST 2016(3)Mon Dec 05 09:25:48 CST 2016(4)Mon Dec 05 09:25:48 CST 2016
java.nio.channels.AsynchronousCloseException
at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:205)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:407)
at com.demo.test.SocketChannelDemo$2.run(SocketChannelDemo.java:118)
at java.lang.Thread.run(Thread.java:745)

socketChannel.connect() = true
serverSocketChannel.accept() != null
service recv : len=31, data=(0)Mon Dec 05 09:26:39 CST 2016
client recv : len=31, data=(0)Mon Dec 05 09:26:39 CST 2016
client send : len=31, data=(0)Mon Dec 05 09:26:39 CST 2016
client send : len=31, data=(1)Mon Dec 05 09:26:39 CST 2016
client recv : len=31, data=(1)Mon Dec 05 09:26:39 CST 2016
service recv : len=31, data=(1)Mon Dec 05 09:26:39 CST 2016
client send : len=31, data=(2)Mon Dec 05 09:26:39 CST 2016
client recv : len=31, data=(2)Mon Dec 05 09:26:39 CST 2016
service recv : len=31, data=(2)Mon Dec 05 09:26:39 CST 2016
client send : len=31, data=(3)Mon Dec 05 09:26:39 CST 2016
service recv : len=31, data=(3)Mon Dec 05 09:26:39 CST 2016
client recv : len=31, data=(3)Mon Dec 05 09:26:39 CST 2016
service recv : len=31, data=(4)Mon Dec 05 09:26:39 CST 2016
client recv : len=31, data=(4)Mon Dec 05 09:26:39 CST 2016
client send : len=31, data=(4)Mon Dec 05 09:26:39 CST 2016
java.nio.channels.AsynchronousCloseException
at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:205)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:407)
at com.demo.test.SocketChannelDemo$2.run(SocketChannelDemo.java:118)
at java.lang.Thread.run(Thread.java:745)