极简、易用、高性能的AIO通信框架

/******************************************************************************* * Copyright (c) 2017-2019, org.smartboot. All rights reserved. * project name: smart-socket * file name: TcpAioSession.java * Date: 2019-12-31 * Author: sandao (zhengjunweimail@163.com) * ******************************************************************************/ package org.smartboot.socket.transport; import org.smartboot.socket.MessageProcessor; import org.smartboot.socket.NetMonitor; import org.smartboot.socket.StateMachineEnum; import org.smartboot.socket.buffer.BufferPage; import org.smartboot.socket.buffer.VirtualBuffer; import org.smartboot.socket.enhance.EnhanceAsynchronousChannelProvider; import java.io.IOException; import java.io.InputStream; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousSocketChannel; import java.nio.channels.CompletionHandler; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.function.Function; /** * AIO传输层会话。 * * <p> * AioSession为smart-socket最核心的类，封装{@link AsynchronousSocketChannel} API接口，简化IO操作。 * </p> * <p> * 其中开放给用户使用的接口为： * <ol> * <li>{@link TcpAioSession#close()}</li> * <li>{@link TcpAioSession#close(boolean)}</li> * <li>{@link TcpAioSession#getAttachment()} </li> * <li>{@link TcpAioSession#getInputStream()} </li> * <li>{@link TcpAioSession#getInputStream(int)} </li> * <li>{@link TcpAioSession#getLocalAddress()} </li> * <li>{@link TcpAioSession#getRemoteAddress()} </li> * <li>{@link TcpAioSession#getSessionID()} </li> * <li>{@link TcpAioSession#isInvalid()} </li> * <li>{@link TcpAioSession#setAttachment(Object)} </li> * </ol> * * </p> * * @author 三刀 * @version V1.0.0 */ final class TcpAioSession extends AioSession { /** * 读事件回调处理 */ private static final CompletionHandler<Integer, TcpAioSession> READ_COMPLETION_HANDLER = new CompletionHandler<Integer, TcpAioSession>() { @Override public void completed(Integer result, TcpAioSession aioSession) { try { aioSession.readCompleted(result); } catch (Throwable throwable) { failed(throwable, aioSession); } } @Override public void failed(Throwable exc, TcpAioSession aioSession) { try { aioSession.config.getProcessor().stateEvent(aioSession, StateMachineEnum.INPUT_EXCEPTION, exc); } catch (Exception e) { e.printStackTrace(); } try { aioSession.close(false); } catch (Exception e) { e.printStackTrace(); } } }; /** * 写事件回调处理 */ private static final CompletionHandler<Integer, TcpAioSession> WRITE_COMPLETION_HANDLER = new CompletionHandler<Integer, TcpAioSession>() { @Override public void completed(Integer result, TcpAioSession aioSession) { try { aioSession.writeCompleted(result); } catch (Throwable throwable) { failed(throwable, aioSession); } } @Override public void failed(Throwable exc, TcpAioSession aioSession) { try { aioSession.config.getProcessor().stateEvent(aioSession, StateMachineEnum.OUTPUT_EXCEPTION, exc); } catch (Exception e) { e.printStackTrace(); } try { aioSession.close(); } catch (Exception e) { e.printStackTrace(); } } }; /** * 底层通信channel对象 */ private final AsynchronousSocketChannel channel; /** * 输出流 */ private final WriteBuffer byteBuf; /** * 输出信号量,防止并发write导致异常 */ private final Semaphore semaphore = new Semaphore(1); /** * 读缓冲。 * <p>大小取决于AioQuickClient/AioQuickServer设置的setReadBufferSize</p> */ private VirtualBuffer readBuffer; /** * 写缓冲 */ private VirtualBuffer writeBuffer; /** * 同步输入流 */ private InputStream inputStream; private final BufferPage bufferPage; private final IoServerConfig config; /** * @param channel Socket通道 */ TcpAioSession(AsynchronousSocketChannel channel, IoServerConfig config, BufferPage bufferPage, Function<BufferPage, VirtualBuffer> function) { this.channel = channel; this.config = config; this.bufferPage = bufferPage; this.function = function; Consumer<WriteBuffer> flushConsumer = var -> { if (!semaphore.tryAcquire()) { return; } TcpAioSession.this.writeBuffer = var.poll(); if (writeBuffer == null) { semaphore.release(); } else { continueWrite(writeBuffer); } }; byteBuf = new WriteBuffer(bufferPage, flushConsumer, config.getWriteBufferSize(), config.getWriteBufferCapacity()); //触发状态机 config.getProcessor().stateEvent(this, StateMachineEnum.NEW_SESSION, null); doRead(); } private final Function<BufferPage, VirtualBuffer> function; void doRead() { this.readBuffer = function.apply(bufferPage); this.readBuffer.buffer().flip(); signalRead(); } /** * 触发AIO的写操作, * <p>需要调用控制同步</p> */ void writeCompleted(int result) { NetMonitor monitor = config.getMonitor(); if (monitor != null) { monitor.afterWrite(this, result); } if (writeBuffer == null) { writeBuffer = byteBuf.pollItem(); } else if (!writeBuffer.buffer().hasRemaining()) { writeBuffer.clean(); writeBuffer = byteBuf.pollItem(); } if (writeBuffer != null) { continueWrite(writeBuffer); return; } semaphore.release(); //此时可能是Closing或Closed状态 if (status != SESSION_STATUS_ENABLED) { close(); } else { //也许此时有新的消息通过write方法添加到writeCacheQueue中 byteBuf.flush(); } } /** * @return 输入流 */ public WriteBuffer writeBuffer() { return byteBuf; } @Override public ByteBuffer readBuffer() { return readBuffer.buffer(); } @Override public void awaitRead() { modCount++; } /** * 是否立即关闭会话 * * @param immediate true:立即关闭,false:响应消息发送完后关闭 */ public synchronized void close(boolean immediate) { //status == SESSION_STATUS_CLOSED说明close方法被重复调用 if (status == SESSION_STATUS_CLOSED) { // System.out.println("ignore, session:" + getSessionID() + " is closed:"); return; } status = immediate ? SESSION_STATUS_CLOSED : SESSION_STATUS_CLOSING; if (immediate) { try { byteBuf.close(); if (readBuffer != null) { readBuffer.clean(); readBuffer = null; } if (writeBuffer != null) { writeBuffer.clean(); writeBuffer = null; } } finally { IOUtil.close(channel); config.getProcessor().stateEvent(this, StateMachineEnum.SESSION_CLOSED, null); } } else if ((writeBuffer == null || !writeBuffer.buffer().hasRemaining()) && byteBuf.isEmpty()) { close(true); } else { config.getProc