1.Java NIO服务端创建
首先，我们通过一个时序图来看下如何创建一个NIO服务端并启动监听，接收多个客户端的连接，进行消息的异步读写。
示例代码(参考文献【2】)：
import java.io.IOException;
import java.net.InetSocketAddress;
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; /**
* User: mihasya
* Date: Jul 25, 2010
* Time: 9:09:03 AM
*/
public class JServer {
public static void main (String[] args) {
ServerSocketChannel sch = null;
Selector sel = null; try {
// setup the socket we're listening for connections on.
InetSocketAddress addr = new InetSocketAddress(8400);
sch = ServerSocketChannel.open();
sch.configureBlocking(false);
sch.socket().bind(addr);
// setup our selector and register the main socket on it
sel = Selector.open();
sch.register(sel, SelectionKey.OP_ACCEPT);
} catch (IOException e) {
System.out.println("Couldn't setup server socket");
System.out.println(e.getMessage());
System.exit(1);
} // fire up the listener thread, pass it our selector
ListenerThread listener = new ListenerThread(sel);
listener.run();
} /*
* the thread is completely unnecessary, it could all just happen
* in main()
*/
class ListenerThread extends Thread {
Selector sel = null;
ListenerThread(Selector sel) {
this.sel = sel;
} public void run() {
while (true) { // our canned response for now
ByteBuffer resp = ByteBuffer.wrap(new String("got it\n").getBytes());
try {
// loop over all the sockets that are ready for some activity
while (this.sel.select() > 0) {
Set keys = this.sel.selectedKeys();
Iterator i = keys.iterator();
while (i.hasNext()) {
SelectionKey key = (SelectionKey)i.next();
if (key.isAcceptable()) {
// this means that a new client has hit the port our main
// socket is listening on, so we need to accept the connection
// and add the new client socket to our select pool for reading
// a command later
System.out.println("Accepting connection!");
// this will be the ServerSocketChannel we initially registered
// with the selector in main()
ServerSocketChannel sch = (ServerSocketChannel)key.channel();
SocketChannel ch = sch.accept();
ch.configureBlocking(false);
ch.register(this.sel, SelectionKey.OP_READ);
} else if (key.isReadable()) {
// one of our client sockets has received a command and
// we're now ready to read it in
System.out.println("Accepting command!");
SocketChannel ch = (SocketChannel)key.channel();
ByteBuffer buf = ByteBuffer.allocate(200);
ch.read(buf);
buf.flip();
Charset charset = Charset.forName("UTF-8");
CharsetDecoder decoder = charset.newDecoder();
CharBuffer cbuf = decoder.decode(buf);
System.out.print(cbuf.toString());
// re-register this socket with the selector, this time
// for writing since we'll want to write something to it
// on the next go-around
ch.register(this.sel, SelectionKey.OP_WRITE);
} else if (key.isWritable()) {
// we are ready to send a response to one of the client sockets
// we had read a command from previously
System.out.println("Sending response!");
SocketChannel ch = (SocketChannel)key.channel();
ch.write(resp);
resp.rewind();
// we may get another command from this guy, so prepare
// to read again. We could also close the channel, but
// that sort of defeats the whole purpose of doing async
ch.register(this.sel, SelectionKey.OP_READ);
}
i.remove();
}
}
} catch (IOException e) {
System.out.println("Error in poll loop");
System.out.println(e.getMessage());
System.exit(1);
}
}
}
}
}
从上面的代码可以看出java nio的通用步骤：
1.打开ServerSocketChannel，用于监听客户端的连接，它是所有客户端连接的父通道，绑定监听端口，设置客户端连接方式为非阻塞模式。
2.打开多路复用器并启动服务端监听线程，将ServerSocketChannel注册到Reactor线程的多路复用器Selector上，监听ACCEPT状态。
3.多路复用器监听到有新的客户端接入，处理新的接入请求，完成TCP三次握手后，与客户端建立物理链路。
2. Netty服务端创建
2.1 打开ServerSocketChannel
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
创建一个ServerSocketChannel的过程：
/**
* Create a new instance
*/
public NioServerSocketChannel() {
this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}
调用newSocket方法：
private static ServerSocketChannel newSocket(SelectorProvider provider) {
try {
/**
* Use the {@link SelectorProvider} to open {@link SocketChannel} and so remove condition in
* {@link SelectorProvider#provider()} which is called by each ServerSocketChannel.open() otherwise.
*
* See <a href="See https://github.com/netty/netty/issues/2308">#2308</a>.
*/
return provider.openServerSocketChannel();
} catch (IOException e) {
throw new ChannelException(
"Failed to open a server socket.", e);
}
}
其中的provider.openServerSocketChannel()就是java nio的实现。设置非阻塞模式包含在父类中：
/**
* Create a new instance
*
* @param parent the parent {@link Channel} by which this instance was created. May be {@code null}
* @param ch the underlying {@link SelectableChannel} on which it operates
* @param readInterestOp the ops to set to receive data from the {@link SelectableChannel}
*/
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
this.ch = ch;
this.readInterestOp = readInterestOp;
try {
ch.configureBlocking(false);
} catch (IOException e) {
try {
ch.close();
} catch (IOException e2) {
if (logger.isWarnEnabled()) {
logger.warn(
"Failed to close a partially initialized socket.", e2);
}
} throw new ChannelException("Failed to enter non-blocking mode.", e);
}
}
2.2 打开多路复用器过程
NioEventLoop负责调度和执行Selector轮询操作，选择准备就绪的Channel集合，相关代码如下：
@Override
protected void run() {
for (;;) {
boolean oldWakenUp = wakenUp.getAndSet(false);
try {
if (hasTasks()) {
selectNow();
} else {
select(oldWakenUp); // 'wakenUp.compareAndSet(false, true)' is always evaluated
// before calling 'selector.wakeup()' to reduce the wake-up
// overhead. (Selector.wakeup() is an expensive operation.)
//
// However, there is a race condition in this approach.
// The race condition is triggered when 'wakenUp' is set to
// true too early.
//
// 'wakenUp' is set to true too early if:
// 1) Selector is waken up between 'wakenUp.set(false)' and
// 'selector.select(...)'. (BAD)
// 2) Selector is waken up between 'selector.select(...)' and
// 'if (wakenUp.get()) { ... }'. (OK)
//
// In the first case, 'wakenUp' is set to true and the
// following 'selector.select(...)' will wake up immediately.
// Until 'wakenUp' is set to false again in the next round,
// 'wakenUp.compareAndSet(false, true)' will fail, and therefore
// any attempt to wake up the Selector will fail, too, causing
// the following 'selector.select(...)' call to block
// unnecessarily.
//
// To fix this problem, we wake up the selector again if wakenUp
// is true immediately after selector.select(...).
// It is inefficient in that it wakes up the selector for both
// the first case (BAD - wake-up required) and the second case
// (OK - no wake-up required). if (wakenUp.get()) {
selector.wakeup();
}
} cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
processSelectedKeys();
runAllTasks();
} else {
final long ioStartTime = System.nanoTime(); processSelectedKeys(); final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
} if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
break;
}
}
} catch (Throwable t) {
logger.warn("Unexpected exception in the selector loop.", t); // Prevent possible consecutive immediate failures that lead to
// excessive CPU consumption.
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// Ignore.
}
}
}
}
2.2.1 绑定处理的key
private void processSelectedKeys() {
if (selectedKeys != null) {
processSelectedKeysOptimized(selectedKeys.flip());
} else {
processSelectedKeysPlain(selector.selectedKeys());
}
}
以processSelectedKeysPlain为例：
private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
// check if the set is empty and if so just return to not create garbage by
// creating a new Iterator every time even if there is nothing to process.
// See https://github.com/netty/netty/issues/597
if (selectedKeys.isEmpty()) {
return;
}
Iterator<SelectionKey> i = selectedKeys.iterator();
for (;;) {
final SelectionKey k = i.next();
final Object a = k.attachment();
i.remove();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
if (!i.hasNext()) {
break;
}
if (needsToSelectAgain) {
selectAgain();
selectedKeys = selector.selectedKeys();
// Create the iterator again to avoid ConcurrentModificationException
if (selectedKeys.isEmpty()) {
break;
} else {
i = selectedKeys.iterator();
}
}
}
}
2.3 绑定端口，接收请求：
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(PORT).sync();
调用bind程序
private ChannelFuture doBind(final SocketAddress localAddress) {
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
if (regFuture.isDone()) {
// At this point we know that the registration was complete and successful.
ChannelPromise promise = channel.newPromise();
doBind0(regFuture, channel, localAddress, promise);
return promise;
} else {
// Registration future is almost always fulfilled already, but just in case it's not.
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Throwable cause = future.cause();
if (cause != null) {
// Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
// IllegalStateException once we try to access the EventLoop of the Channel.
promise.setFailure(cause);
} else {
// Registration was successful, so set the correct executor to use.
// See https://github.com/netty/netty/issues/2586
promise.executor = channel.eventLoop();
}
doBind0(regFuture, channel, localAddress, promise);
}
});
return promise;
}
}
最终的绑定由dobind0来完成
private static void doBind0(
final ChannelFuture regFuture, final Channel channel,
final SocketAddress localAddress, final ChannelPromise promise) { // This method is invoked before channelRegistered() is triggered. Give user handlers a chance to set up
// the pipeline in its channelRegistered() implementation.
channel.eventLoop().execute(new OneTimeTask() {
@Override
public void run() {
if (regFuture.isSuccess()) {
channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
} else {
promise.setFailure(regFuture.cause());
}
}
});
}
具体实现：
private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
// close the channel if the key is not valid anymore
unsafe.close(unsafe.voidPromise());
return;
}
try {
int readyOps = k.readyOps();
// Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
// to a spin loop
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
if (!ch.isOpen()) {
// Connection already closed - no need to handle write.
return;
}
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
ch.unsafe().forceFlush();
}
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
// See https://github.com/netty/netty/issues/924
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
参考文献
【1】http://www.infoq.com/cn/articles/netty-server-create
【2】https://github.com/mihasya/sample-java-nio-server/blob/master/src/JServer.java