public interface Protocol<T> {
    public T decode(final ByteBuffer data, AioSession<T> session, boolean eof);
    public ByteBuffer encode(T msg, AioSession<T> session);
}复制代码

public interface MessageProcessor<T> {
    public void process(AioSession<T> session, T msg);
    void stateEvent(AioSession<T> session, @StateMachine int stateMachine, Throwable throwable);
}复制代码

public interface Filter<T> {
    public void readFilter(AioSession<T> session, int readSize);
    public void processFilter(AioSession<T> session, T msg);
    public void processFailHandler(AioSession<T> session, T msg, Throwable e);
    public void writeFilter(AioSession<T> session, int writeSize);
}复制代码

public final void write(final ByteBuffer buffer) throws IOException {
    if (isInvalid()) {
        throw new IOException("session is " + (status == SESSION_STATUS_CLOSED ? "closed" : "invalid"));
    if (!buffer.hasRemaining()) {
        throw new InvalidObjectException("buffer has no remaining");
    //是否设置了输出队列
    if (ioServerConfig.getWriteQueueSize() <= 0) {
        try {
            semaphore.acquire();
            //若当前无待输出的数据，则立即输出buffer
            writeBuffer = buffer;
            writeToChannel0(writeBuffer);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IOException(e.getMessage());
        return;
    } else if ((semaphore.tryAcquire())) {
    //若当前无待输出的数据，则立即输出buffer
        writeBuffer = buffer;
          writeToChannel0(writeBuffer);
        return;
    try {
        //存放到写队列中
        writeCacheQueue.put(buffer);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    // 因为是异步的所以这里再次判断是否空闲
    if (semaphore.tryAcquire()) {
        writeToChannel();
}复制代码

void writeToChannel() {
    if (writeBuffer != null && writeBuffer.hasRemaining()) {
          writeToChannel0(writeBuffer);
        return;
    if (writeCacheQueue == null || writeCacheQueue.size() == 0) {
        writeBuffer = null;
        semaphore.release();
        //此时可能是Closing或Closed状态
        if (isInvalid()) {
            close();
        //也许此时有新的消息通过write方法添加到writeCacheQueue中
        else if (writeCacheQueue != null && writeCacheQueue.size() > 0 && semaphore.tryAcquire()) {
            writeToChannel();
        return;
    int totalSize = writeCacheQueue.expectRemaining(MAX_WRITE_SIZE);
    ByteBuffer headBuffer = writeCacheQueue.poll();
    if (headBuffer.remaining() == totalSize) {
        writeBuffer = headBuffer;
    } else {
        if (writeBuffer == null || totalSize << 1 <= writeBuffer.capacity() || totalSize > writeBuffer.capacity()) {
            writeBuffer = ByteBuffer.allocate(totalSize);
        } else {
            writeBuffer.clear().limit(totalSize);
        writeBuffer.put(headBuffer);
        writeCacheQueue.pollInto(writeBuffer);
        writeBuffer.flip();
    writeToChannel0(writeBuffer);
}复制代码

protected final void writeToChannel0(ByteBuffer buffer) {
    if (null != channel && channel.isOpen()) {
        channel.write(buffer, this, writeCompletionHandler);
}复制代码

void readFromChannel(Integer readSize) {
    readBuffer.flip();
    T dataEntry;
    while ((dataEntry = ioServerConfig.getProtocol().decode(readBuffer, this, readSize)) != null) {
        //处理消息
        try {
            for (Filter<T> h : ioServerConfig.getFilters()) {
                h.processFilter(this, dataEntry);
            ioServerConfig.getProcessor().process(this, dataEntry);
        } catch (Exception e) {
            ioServerConfig.getProcessor().stateEvent(this, StateMachine.PROCESS_EXCEPTION, e);
            for (Filter<T> h : ioServerConfig.getFilters()) {
                h.processFail(this, dataEntry, e);
    if (readSize == -1 || status == SESSION_STATUS_CLOSING) {
        close(false);
        return;
    if (status == SESSION_STATUS_CLOSED) {
        return;
    //数据读取完毕
    if (readBuffer.remaining() == 0) {
        readBuffer.clear();
    } else if (readBuffer.position() > 0) {
        // 仅当发生数据读取时调用compact,减少内存拷贝
        readBuffer.compact();
    } else {
        readBuffer.position(readBuffer.limit());
        readBuffer.limit(readBuffer.capacity());
    readFromChannel0(readBuffer);
}复制代码

protected final void readFromChannel0(ByteBuffer buffer) {
    if (null != channel && channel.isOpen()) {
        channel.read(buffer, this, readCompletionHandler);
}复制代码

public AioQuickClient(String host, int port, Protocol<T> protocol, MessageProcessor<T> messageProcessor) {
    mConfig.setHost(host);
    mConfig.setPort(port);
    mConfig.setProtocol(protocol);
    mConfig.setProcessor(messageProcessor);
}复制代码

public final void start() {
    try {
        mSelector = Selector.open();
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.configureBlocking(false);
        SelectionKey selectionKey = socketChannel.register(mSelector, SelectionKey.OP_CONNECT);
        mAsynchronousSocketChannel = new AsynchronousSocketChannel(selectionKey);
        socketChannel.connect(new InetSocketAddress(mConfig.getHost(), mConfig.getPort()));
//            ToolLog.e(TAG, "开始连接服务器");
        ClientThread serverThread = new ClientThread();
        serverThread.start();
    } catch (final IOException | UnresolvedAddressException e) {
        e.printStackTrace();
        mConfig.getProcessor().stateEvent(mSession, StateMachine.SESSION_CONNECT_FAILED, e);
}复制代码

    @Override
    public void run() {
        try {
            while (mAsynchronousSocketChannel.isOpen()) {
                // 优先获取SelectionKey,若无关注事件触发则阻塞在selector.select(),减少select被调用次数
                Set<SelectionKey> keySet = mSelector.selectedKeys();
                if (keySet.isEmpty()) {
                    mSelector.select();
                Iterator<SelectionKey> keyIterator = keySet.iterator();
                // 执行本次已触发待处理的事件
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    // 读取数据
                    if (key.isReadable()) {
                        mAsynchronousSocketChannel.removeOps(SelectionKey.OP_READ);
                        mAsynchronousSocketChannel.doRead();
                    } else if (key.isWritable()) {// 输出数据
                        mAsynchronousSocketChannel.removeOps(SelectionKey.OP_WRITE);
                        mAsynchronousSocketChannel.doWrite();
                    } else if (key.isConnectable()) {// 建立新连接,Client触发Connect,Server触发Accept
                        acceptConnect(key);
                    // 移除已处理的事件
                    keyIterator.remove();
        } catch (Exception e) {
            shutdown();
}复制代码

private void acceptConnect(SelectionKey key) throws IOException {
    SocketChannel channel = (SocketChannel) key.channel();
    channel.finishConnect();
    key.interestOps(key.interestOps() & ~SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
     // 创建AioSession会话
    mSession = new AioSession<>(mAsynchronousSocketChannel, mConfig, new ReadCompletionHandler(), new WriteCompletionHandler());
    mSession.initSession();
    //触发当前已连接状态
    mConfig.getProcessor().stateEvent(mSession, StateMachine.NEW_SESSION, null);
}复制代码