开篇
回想研究生期间在H3C做项目的时候第一次接触epoll的异步事件，心血来潮看了下java的NIO的实现，希望同样感兴趣的人一起看看。Netty是java NIO的集大成者，一定要看看。
java NIO server demo
socket server端工作标准流程
- 创建socket： 创建ServerSocketChannel，通过ServerSocketChannel.open()方法。
- 绑定socket：ServerSocketChannel绑定端口，通过serverSocketChannel.bind()方法。
- 前置准备： 创建selector对象，通过Selector.open()方法。
- 前置准备： 注册Channel到selector并绑定事件，通过serverSocketChannel.register()。
- 监听端口号： 通过listen()方法开始进入监听。
- 处理事件： while循环中等待select操作返回区分连接还是数据进行不同处理。
public class NIOServer {
private Selector selector;
public void initServer(int port) throws IOException {
// 获得一个ServerSocketChannel通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 设置通道为非阻塞
serverSocketChannel.configureBlocking(false);
// 将该通道对应的ServerSocket绑定到port端口
serverSocketChannel.bind(new InetSocketAddress(port));
// 获得一个通道管理器
this.selector = Selector.open();
// 将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后，
// 当该事件到达时，selector.select()会返回，如果该事件没到达selector.select()会一直阻塞。
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
}
public void listen() throws IOException {
System.out.println("服务端启动成功！");
// 轮询访问selector
while (true) {
// 当注册的事件到达时，方法返回；否则,该方法会一直阻塞
selector.select();
// 获得selector中选中的项的迭代器，选中的项为注册的事件
Iterator<SelectionKey> ite = this.selector.selectedKeys().iterator();
while (ite.hasNext()) {
SelectionKey key = (SelectionKey) ite.next();
// 删除已选的key,以防重复处理
ite.remove();
if (key.isAcceptable()) {// 客户端请求连接事件
ServerSocketChannel server = (ServerSocketChannel) key.channel();
// 获得和客户端连接的通道
SocketChannel channel = server.accept();
// 设置成非阻塞
channel.configureBlocking(false);
// 在这里可以给客户端发送信息哦
channel.write(ByteBuffer.wrap(new String("向客户端发送了一条信息")
.getBytes("utf-8")));
// 在和客户端连接成功之后，为了可以接收到客户端的信息，需要给通道设置读的权限。
channel.register(this.selector, SelectionKey.OP_READ);
} else if (key.isReadable()) {// 获得了可读的事件
read(key);
}
}
}
}
public void read(SelectionKey key) throws IOException {
// 服务器可读取消息:得到事件发生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
// 创建读取的缓冲区
ByteBuffer buffer = ByteBuffer.allocate(512);
channel.read(buffer);
byte[] data = buffer.array();
String msg = new String(data).trim();
System.out.println("服务端收到信息：" + msg);
ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes("utf-8"));
channel.write(outBuffer);// 将消息回送给客户端
}
public static void main(String[] args) throws IOException {
NIOServer server = new NIOServer();
server.initServer(8000);
server.listen();
}
}
ServerSocketChannel和Selector初始化过程
在java NIO Server的标准过程中，有两个核心的操作需要深入分析一下，分别是ServerSocketChannel.open() 和 Selector.open()两个过程，这里针对这两个对象的初始化流程进行下细致的分解。
// ServerSocketChannel的初始化过程
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// Selector的初始化过程
selector = Selector.open();
通用逻辑抽取
- ServerSocketChannel.open()=SelectorProvider.provider().openServerSocketChannel()
- Selector.open()=SelectorProvider.provider().openSelector()
两者有共同点在于都调用了SelectorProvider.provider()方法，所以先把相同部分进行分析。
public abstract class ServerSocketChannel extends AbstractSelectableChannel
implements NetworkChannel
{
protected ServerSocketChannel(SelectorProvider provider) {
super(provider);
}
public static ServerSocketChannel open() throws IOException {
return SelectorProvider.provider().openServerSocketChannel();
}
}
public abstract class Selector implements Closeable {
protected Selector() { }
public static Selector open() throws IOException {
return SelectorProvider.provider().openSelector();
}
}
SelectorProvider对象创建
- SelectorProvider.provider()方法会在内部创建唯一的SelectorProvider对象，通过锁来保证创建唯一对象。
- SelectorProvider对象通过DefaultSelectorProvider.create()方法进行创建。
- DefaultSelectorProvider.create()方法内部根据实际系统创建不同的对象，以linux环境中EPollSelectorProvider对象为例继续分析。
public abstract class SelectorProvider {
private static final Object lock = new Object();
private static SelectorProvider provider = null;
public static SelectorProvider provider() {
synchronized (lock) {
if (provider != null)
return provider;
return AccessController.doPrivileged(
new PrivilegedAction<SelectorProvider>() {
public SelectorProvider run() {
if (loadProviderFromProperty())
return provider;
if (loadProviderAsService())
return provider;
provider = sun.nio.ch.DefaultSelectorProvider.create();
return provider;
}
});
}
}
}
public class DefaultSelectorProvider {
public static SelectorProvider create() {
String osname = AccessController.doPrivileged(
new GetPropertyAction("os.name"));
if ("SunOS".equals(osname)) {
return new sun.nio.ch.DevPollSelectorProvider();
}
// use EPollSelectorProvider for Linux kernels >= 2.6
if ("Linux".equals(osname)) {
String osversion = AccessController.doPrivileged(
new GetPropertyAction("os.version"));
String[] vers = osversion.split("\\.", 0);
if (vers.length >= 2) {
try {
int major = Integer.parseInt(vers[0]);
int minor = Integer.parseInt(vers[1]);
if (major > 2 || (major == 2 && minor >= 6)) {
return new sun.nio.ch.EPollSelectorProvider();
}
} catch (NumberFormatException x) {
// format not recognized
}
}
}
return new sun.nio.ch.PollSelectorProvider();
}
}
EPollSelectorProvider的操作过程
- SelectorProvider.provider().openServerSocketChannel()调用EPollSelectorProvider的openServerSocketChannel()方法返回EPollSelectorImpl对象。
- SelectorProvider.provider().openSelector()调用EPollSelectorProvider的openSelector()方法返回ServerSocketChannelImpl对象。
继续分析ServerSocketChannelImpl对象和EPollSelectorImpl对象。
public class EPollSelectorProvider extends SelectorProviderImpl
{
public AbstractSelector openSelector() throws IOException {
return new EPollSelectorImpl(this);
}
public Channel inheritedChannel() throws IOException {
return InheritedChannel.getChannel();
}
}
public abstract class SelectorProviderImpl extends SelectorProvider
{
public DatagramChannel openDatagramChannel() throws IOException {
return new DatagramChannelImpl(this);
}
public DatagramChannel openDatagramChannel(ProtocolFamily family) throws IOException {
return new DatagramChannelImpl(this, family);
}
public Pipe openPipe() throws IOException {
return new PipeImpl(this);
}
public abstract AbstractSelector openSelector() throws IOException;
public ServerSocketChannel openServerSocketChannel() throws IOException {
return new ServerSocketChannelImpl(this);
}
public SocketChannel openSocketChannel() throws IOException {
return new SocketChannelImpl(this);
}
}
EPollSelectorImpl对象
- EPollSelectorImpl构造函数创建内部通信的socket对IOUtil.makePipe(false)。
- EPollSelectorImpl的doSelect方法负责返回事件到来的fds。
- EPollSelectorImpl的fdToKey的map保存fd和SelectionKey的映射。
class EPollSelectorImpl extends SelectorImpl
{
// File descriptors used for interrupt
protected int fd0;
protected int fd1;
// The poll object
EPollArrayWrapper pollWrapper;
// Maps from file descriptors to keys
private Map<Integer,SelectionKeyImpl> fdToKey;
// True if this Selector has been closed
private volatile boolean closed = false;
// Lock for interrupt triggering and clearing
private Object interruptLock = new Object();
private boolean interruptTriggered = false;
EPollSelectorImpl(SelectorProvider sp) {
super(sp);
long pipeFds = IOUtil.makePipe(false);
fd0 = (int) (pipeFds >>> 32);
fd1 = (int) pipeFds;
pollWrapper = new EPollArrayWrapper();
pollWrapper.initInterrupt(fd0, fd1);
fdToKey = new HashMap<Integer,SelectionKeyImpl>();
}
protected int doSelect(long timeout) throws IOException
{
if (closed)
throw new ClosedSelectorException();
processDeregisterQueue();
try {
begin();
// 等待事件到来，收集事件到来的socket的fd并用来处理
pollWrapper.poll(timeout);
} finally {
end();
}
processDeregisterQueue();
// 更新需要写入的keys
int numKeysUpdated = updateSelectedKeys();
if (pollWrapper.interrupted()) {
// Clear the wakeup pipe
pollWrapper.putEventOps(pollWrapper.interruptedIndex(), 0);
synchronized (interruptLock) {
pollWrapper.clearInterrupted();
IOUtil.drain(fd0);
interruptTriggered = false;
}
}
return numKeysUpdated;
}
private int updateSelectedKeys() {
int entries = pollWrapper.updated;
int numKeysUpdated = 0;
for (int i=0; i<entries; i++) {
int nextFD = pollWrapper.getDescriptor(i);
SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
// ski is null in the case of an interrupt
if (ski != null) {
int rOps = pollWrapper.getEventOps(i);
if (selectedKeys.contains(ski)) {
if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
numKeysUpdated++;
}
} else {
ski.channel.translateAndSetReadyOps(rOps, ski);
if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
// selectedKeys保存ski也就是事件到的socket连接
// ski的对象数据结构需要好好研究一下
selectedKeys.add(ski);
numKeysUpdated++;
}
}
}
}
return numKeysUpdated;
}
}
ServerSocketChannelImpl对象
- ServerSocketChannelImpl extends ServerSocketChannel
- ServerSocketChannel extends AbstractSelectableChannel
- ServerSocketChannelImpl对象提供bind()&accept()方法
- ServerSocketChannelImpl的accept方法内部创建新连接的SocketChannelImpl对象返回
class ServerSocketChannelImpl extends ServerSocketChannel implements SelChImpl
{
private final Object stateLock = new Object();
private SocketAddress localAddress;
ServerSocket socket;
ServerSocketChannelImpl(SelectorProvider sp) throws IOException {
super(sp);
this.fd = Net.serverSocket(true);
this.fdVal = IOUtil.fdVal(fd);
this.state = ST_INUSE;
}
ServerSocketChannelImpl(SelectorProvider sp,
FileDescriptor fd,
boolean bound) throws IOException
{
super(sp);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
this.state = ST_INUSE;
if (bound)
localAddress = Net.localAddress(fd);
}
@Override
public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
// 省略相关代码
}
public SocketChannel accept() throws IOException {
// 省略相关代码
}
}
public abstract class AbstractSelectableChannel extends SelectableChannel
{
protected AbstractSelectableChannel(SelectorProvider provider) {
this.provider = provider;
}
public final SelectionKey register(Selector sel, int ops,
Object att)
throws ClosedChannelException
{
synchronized (regLock) {
if (!isOpen())
throw new ClosedChannelException();
if ((ops & ~validOps()) != 0)
throw new IllegalArgumentException();
if (blocking)
throw new IllegalBlockingModeException();
SelectionKey k = findKey(sel);
if (k != null) {
k.interestOps(ops);
k.attach(att);
}
if (k == null) {
// New registration
synchronized (keyLock) {
if (!isOpen())
throw new ClosedChannelException();
k = ((AbstractSelector)sel).register(this, ops, att);
addKey(k);
}
}
return k;
}
}
}
select过程
select执行过程
- 执行selector.select()操作时实际是调用了子类实现的doSelect()方法。
- 进一步跟进子类的doSelect()方法。
abstract class SelectorImpl extends AbstractSelector
{
// 保存事件到来的keys
protected Set<SelectionKey> selectedKeys;
protected HashSet<SelectionKey> keys;
private Set<SelectionKey> publicKeys; // Immutable
private Set<SelectionKey> publicSelectedKeys; // Removal allowed, but not addition
protected abstract int doSelect(long timeout) throws IOException;
private int lockAndDoSelect(long timeout) throws IOException {
synchronized (this) {
if (!isOpen())
throw new ClosedSelectorException();
synchronized (publicKeys) {
synchronized (publicSelectedKeys) {
return doSelect(timeout);
}
}
}
}
public int select(long timeout) throws IOException
{
if (timeout < 0)
throw new IllegalArgumentException("Negative timeout");
return lockAndDoSelect((timeout == 0) ? -1 : timeout);
}
public int select() throws IOException {
return select(0);
}
public Set<SelectionKey> selectedKeys() {
if (!isOpen() && !Util.atBugLevel("1.4"))
throw new ClosedSelectorException();
return publicSelectedKeys;
}
}
- pollWrapper.poll(timeout)以超时等待的形式等待epoll的消息通知。
- 通过updateSelectedKeys方法收集有事件到达的fds保存到selectedKeys。
class EPollSelectorImpl extends SelectorImpl
{
protected int doSelect(long timeout) throws IOException
{
if (closed)
throw new ClosedSelectorException();
processDeregisterQueue();
try {
begin();
// 等待事件到来，收集事件到来的socket的fd并用来处理
pollWrapper.poll(timeout);
} finally {
end();
}
processDeregisterQueue();
// 更新需要写入的keys
int numKeysUpdated = updateSelectedKeys();
if (pollWrapper.interrupted()) {
// Clear the wakeup pipe
pollWrapper.putEventOps(pollWrapper.interruptedIndex(), 0);
synchronized (interruptLock) {
pollWrapper.clearInterrupted();
IOUtil.drain(fd0);
interruptTriggered = false;
}
}
return numKeysUpdated;
}
private int updateSelectedKeys() {
int entries = pollWrapper.updated;
int numKeysUpdated = 0;
for (int i=0; i<entries; i++) {
int nextFD = pollWrapper.getDescriptor(i);
SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
// ski is null in the case of an interrupt
if (ski != null) {
int rOps = pollWrapper.getEventOps(i);
if (selectedKeys.contains(ski)) {
if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
numKeysUpdated++;
}
} else {
ski.channel.translateAndSetReadyOps(rOps, ski);
if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
// selectedKeys保存ski也就是事件到的socket连接
// ski的对象数据结构需要好好研究一下
selectedKeys.add(ski);
numKeysUpdated++;
}
}
}
}
return numKeysUpdated;
}
}
accept过程
- accept的过程很简单就是accept新socket并创建SocketChannelImpl返回即可。
- SocketChannelImpl对象后面需要注册到Selector当中所以需要进一步分析。
public SocketChannel accept() throws IOException {
// 省略相关代码
try {
// 省略相关代码
// 新accept的socket放在newfd当中
n = accept0(this.fd, newfd, isaa);
}
}
IOUtil.configureBlocking(newfd, true);
InetSocketAddress isa = isaa[0];
// 通过SocketChannelImpl包装newfd对象
sc = new SocketChannelImpl(provider(), newfd, isa);
// 省略相关代码
return sc;
}
}
SocketChannelImpl对象
- SocketChannelImpl可以理解为普通Socket的封装，包括read/write等方法
- SocketChannelImpl extends SocketChannel extends AbstractSelectableChannel
- AbstractSelectableChannel提供register到selector对象的方法
class SocketChannelImpl extends SocketChannel implements SelChImpl
{
SocketChannelImpl(SelectorProvider sp) throws IOException {
super(sp);
this.fd = Net.socket(true);
this.fdVal = IOUtil.fdVal(fd);
this.state = ST_UNCONNECTED;
}
SocketChannelImpl(SelectorProvider sp,
FileDescriptor fd,
boolean bound)
throws IOException
{
super(sp);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
this.state = ST_UNCONNECTED;
if (bound)
this.localAddress = Net.localAddress(fd);
}
SocketChannelImpl(SelectorProvider sp,
FileDescriptor fd, InetSocketAddress remote)
throws IOException
{
super(sp);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
this.state = ST_CONNECTED;
this.localAddress = Net.localAddress(fd);
this.remoteAddress = remote;
}
public long read(ByteBuffer[] dsts, int offset, int length)
throws IOException
{
// 读数据的逻辑
}
public int write(ByteBuffer buf) throws IOException {
// 写数据的逻辑
}
}
register过程
- register过程并没有调用epollCtl方法添加fd到selector当中
- register过程真正是保存fd到待绑定的列表当中
- 在SelectorImpl中执行pollWrapper.poll(timeout)方法先把fd列表执行epollCtl添加selector当中，在通过epollWait获取事件到来
public abstract class AbstractSelectableChannel extends SelectableChannel
{
public final SelectionKey register(Selector sel, int ops,
Object att)
throws ClosedChannelException
{
synchronized (regLock) {
if (!isOpen())
throw new ClosedChannelException();
if ((ops & ~validOps()) != 0)
throw new IllegalArgumentException();
if (blocking)
throw new IllegalBlockingModeException();
SelectionKey k = findKey(sel);
if (k != null) {
k.interestOps(ops);
k.attach(att);
}
if (k == null) {
// New registration
synchronized (keyLock) {
if (!isOpen())
throw new ClosedChannelException();
k = ((AbstractSelector)sel).register(this, ops, att);
addKey(k);
}
}
return k;
}
}
}
abstract class SelectorImpl extends AbstractSelector
{
protected final SelectionKey register(AbstractSelectableChannel ch,
int ops,
Object attachment)
{
if (!(ch instanceof SelChImpl))
throw new IllegalSelectorException();
SelectionKeyImpl k = new SelectionKeyImpl((SelChImpl)ch, this);
k.attach(attachment);
synchronized (publicKeys) {
implRegister(k);
}
k.interestOps(ops);
return k;
}
}
abstract class AbstractPollSelectorImpl extends SelectorImpl
{
protected void implRegister(SelectionKeyImpl ski) {
synchronized (closeLock) {
if (closed)
throw new ClosedSelectorException();
// Check to see if the array is large enough
if (channelArray.length == totalChannels) {
// Make a larger array
int newSize = pollWrapper.totalChannels * 2;
SelectionKeyImpl temp[] = new SelectionKeyImpl[newSize];
// Copy over
for (int i=channelOffset; i<totalChannels; i++)
temp[i] = channelArray[i];
channelArray = temp;
// Grow the NativeObject poll array
pollWrapper.grow(newSize);
}
channelArray[totalChannels] = ski;
ski.setIndex(totalChannels);
// 核心的将channel添加到pollWrapper当中
pollWrapper.addEntry(ski.channel);
totalChannels++;
keys.add(ski);
}
}
}
class EPollArrayWrapper {
int poll(long timeout) throws IOException {
updateRegistrations();
updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd);
for (int i=0; i<updated; i++) {
if (getDescriptor(i) == incomingInterruptFD) {
interruptedIndex = i;
interrupted = true;
break;
}
}
return updated;
}
void updateRegistrations() {
synchronized (updateList) {
Updator u = null;
while ((u = updateList.poll()) != null) {
SelChImpl ch = u.channel;
if (!ch.isOpen())
continue;
// if the events are 0 then file descriptor is put into "idle
// set" to prevent it being polled
if (u.events == 0) {
boolean added = idleSet.add(u.channel);
// if added to idle set then remove from epoll if registered
if (added && (u.opcode == EPOLL_CTL_MOD))
epollCtl(epfd, EPOLL_CTL_DEL, ch.getFDVal(), 0);
} else {
// events are specified. If file descriptor was in idle set
// it must be re-registered (by converting opcode to ADD)
boolean idle = false;
if (!idleSet.isEmpty())
idle = idleSet.remove(u.channel);
int opcode = (idle) ? EPOLL_CTL_ADD : u.opcode;
epollCtl(epfd, opcode, ch.getFDVal(), u.events);
}
}
}
}
}
