Tio Boot Docs

Appearance

深入解析 Tio 源码：构建高性能 Java 网络应用 ​

在现代网络应用中，处理大量并发连接和高效的数据传输是至关重要的。Tio（简称 t-io）作为一个基于 Java 的高性能网络通信框架，旨在简化这一过程。本文将基于提供的代码片段，深入解析 t-io 的源码结构和关键技术细节，帮助开发者更好地理解并应用这一框架。

目录 ​

  1. Tio 框架概述
  2. 核心配置类：TioConfig
  3. 服务器配置：ServerTioConfig
  4. 服务器启动与监听：TioServer
  5. 连接接收处理：AcceptCompletionHandler
  6. 数据读取与解码：ReadCompletionHandler 与 DecodeTask
  7. 消息处理：HandlePacketTask
  8. 网络通信管理：Tio
  9. 异常处理与连接关闭：CloseTask
  10. 总结与技术要点

Tio 框架概述 ​

Tio 是一个基于 Java AIO 的网络通信框架，旨在提供高性能、易用性和可扩展性。它支持 TCP、WebSocket 等协议，具备心跳机制、IP 黑名单、群组管理等功能。通过抽象化的配置和事件驱动的处理方式，Tio 简化了网络应用的开发过程。

核心配置类：TioConfig ​

TioConfig是 Tio 框架的核心配置类，负责管理所有与网络通信相关的配置和状态。以下是对其关键部分的详细解析。

类结构与成员变量 ​

java
public abstract class TioConfig extends MapWithLockPropSupport {
    static Logger log = LoggerFactory.getLogger(TioConfig.class);

    public static final Set<ServerTioConfig> ALL_SERVER_GROUPCONTEXTS = new HashSet<>();
    public static final Set<ClientTioConfig> ALL_CLIENT_GROUPCONTEXTS = new HashSet<>();
    public static final Set<TioConfig> ALL_GROUPCONTEXTS = new HashSet<>();

    public static final int READ_BUFFER_SIZE = Integer.getInteger("tio.default.read.buffer.size", 20480);
    private final static AtomicInteger ID_ATOMIC = new AtomicInteger();

    private ByteOrder byteOrder = ByteOrder.BIG_ENDIAN;
    public boolean isShortConnection = false;
    public SslConfig sslConfig = null;
    public boolean debug = false;
    public GroupStat groupStat = null;
    public boolean statOn = true;
    public PacketConverter packetConverter = null;

    private CacheFactory cacheFactory;
    @SuppressWarnings("rawtypes")
    private RemovalListenerWrapper ipRemovalListenerWrapper;
    public long startTime = SystemTimer.currTime;
    public long heartbeatTimeout = 1000 * 120;
    public boolean logWhenDecodeError = false;
    private int readBufferSize = READ_BUFFER_SIZE;
    private GroupListener groupListener = null;
    private AioId tioUuid = new DefaultTAioId();

    public ClientNodes clientNodes = new ClientNodes();
    public SetWithLock<ChannelContext> connections = new SetWithLock<ChannelContext>(new HashSet<ChannelContext>());
    public Groups groups = new Groups();
    public Users users = new Users();
    public Tokens tokens = new Tokens();
    public Ids ids = new Ids();
    public BsIds bsIds = new BsIds();
    public Ips ips = new Ips();
    public IpStats ipStats = new IpStats(this, null);;
    protected String id;
    protected int maxDecodeErrorCountForIp = 10;
    protected String name = "Untitled";
    private IpStatListener ipStatListener = DefaultIpStatListener.me;
    private boolean isStopped = false;
    public IpBlacklist ipBlacklist = null;
    public MapWithLock<Integer, Packet> waitingResps = new MapWithLock<Integer, Packet>(new HashMap<Integer, Packet>());

    public boolean disgnostic = EnvUtils.getBoolean(TioCoreConfigKeys.TIO_CORE_DIAGNOSTIC);

    // 构造函数与初始化方法略
}

主要成员变量解析 ​

  1. 连接管理

    • clientNodes：管理所有客户端节点。
    • connections：当前所有的连接集合。
    • groups：管理群组信息。
    • users、tokens、ids、bsIds、ips：分别管理用户、令牌、ID、业务 ID、IP 相关的信息。

  2. 缓存与统计

    • cacheFactory：用于创建缓存工厂，支持不同类型的缓存实现。
    • ipStats：IP 统计信息，支持多种统计时段。
    • groupStat：群组统计信息。
    • heartbeatTimeout：心跳超时时间，默认 120 秒。

  3. 配置与状态

    • byteOrder：字节序，默认大端。
    • sslConfig：SSL 配置，用于启用加密通信。
    • debug：调试模式开关。
    • statOn：统计开关。
    • packetConverter：自定义包转换器。
    • ipBlacklist：IP 黑名单管理。

  4. 生命周期管理

    • startTime：配置启动时间。
    • isStopped：停止标志。
    • id：唯一标识符，通过AtomicInteger递增生成。
    • name：配置名称。

关键方法解析 ​

  1. 初始化方法

    init()方法负责初始化配置，注册到全局上下文集合，并设置默认的 IP 移除监听器。

    java
    public void init() {
    if (cacheFactory == null) {
        this.cacheFactory = ConcurrentMapCacheFactory.INSTANCE;
    }

    if (ipRemovalListenerWrapper == null) {
        setDefaultIpRemovalListenerWrapper();
    }

    ALL_GROUPCONTEXTS.add(this);
    if (this instanceof ServerTioConfig) {
        ALL_SERVER_GROUPCONTEXTS.add((ServerTioConfig) this);
    } else {
        ALL_CLIENT_GROUPCONTEXTS.add((ClientTioConfig) this);
    }

    if (ALL_GROUPCONTEXTS.size() > 20) {
        log.warn("You have created {} TioConfig objects, you might be misusing t-io.", ALL_GROUPCONTEXTS.size());
    }
    this.id = ID_ATOMIC.incrementAndGet() + "";

    if (this.ipStats == null) {
        this.ipStats = new IpStats(this, null);
    }
}

  2. 设置默认 IP 移除监听器

    如果未指定 IP 移除监听器，则设置默认的监听器，用于处理 IP 被移除后的逻辑。

    java
    @SuppressWarnings({ "rawtypes", "unchecked" })
public void setDefaultIpRemovalListenerWrapper() {
    this.ipRemovalListenerWrapper = new RemovalListenerWrapper();
    IpStatMapCacheRemovalListener ipStatMapCacheRemovalListener = new IpStatMapCacheRemovalListener(this, ipStatListener);
    ipRemovalListenerWrapper.setListener(ipStatMapCacheRemovalListener);
}

  3. 抽象方法

    TioConfig是一个抽象类，需要子类实现以下两个方法：

    java
    public abstract AioHandler getAioHandler();
public abstract AioListener getAioListener();
public abstract boolean isServer();
    • getAioHandler()：获取 AIO 处理器，用于处理数据的编码和解码。
    • getAioListener()：获取 AIO 监听器，用于监听连接事件。
    • isServer()：判断是否为服务器配置。

服务器配置：ServerTioConfig ​

ServerTioConfig是TioConfig的一个具体实现，专门用于服务器端的配置管理。

类结构与成员变量 ​

java
public class ServerTioConfig extends TioConfig {
    static Logger log = LoggerFactory.getLogger(ServerTioConfig.class);

    private ServerAioHandler serverAioHandler = null;
    private ServerAioListener serverAioListener = null;
    private Thread checkHeartbeatThread = null;
    private boolean needCheckHeartbeat = true;
    private boolean isShared = false;

    // 构造函数与其他方法略
}

主要成员变量解析 ​

  1. AIO 处理器与监听器

    • serverAioHandler：服务器端 AIO 处理器，负责数据的编码和解码。
    • serverAioListener：服务器端 AIO 监听器，监听连接事件和状态变化。

  2. 心跳检测

    • checkHeartbeatThread：用于定时检查连接的心跳线程。
    • needCheckHeartbeat：是否需要进行心跳检测的标志。

  3. 共享配置

    • isShared：标志配置是否被共享，影响心跳检测的逻辑。

关键方法解析 ​

  1. 使用 SSL

    useSsl方法用于配置 SSL，加密通信数据。

    java
    public void useSsl(String keyStoreFile, String trustStoreFile, String keyStorePwd) throws Exception {
    if (StrUtil.isNotBlank(keyStoreFile) && StrUtil.isNotBlank(trustStoreFile)) {
        SslConfig sslConfig = SslConfig.forServer(keyStoreFile, trustStoreFile, keyStorePwd);
        this.setSslConfig(sslConfig);
    }
}

public void useSsl(InputStream keyStoreInputStream, InputStream trustStoreInputStream, String passwd) throws Exception {
    SslConfig sslConfig = SslConfig.forServer(keyStoreInputStream, trustStoreInputStream, passwd);
    this.setSslConfig(sslConfig);
}

  2. 初始化方法

    init()方法不仅初始化基础配置，还启动心跳检测线程。

    java
    public void init() {
    super.init();
    this.groupStat = new ServerGroupStat();
    GlobalIpBlacklist.INSTANCE.init(this);
    Runnable check = new Runnable() {
        @Override
        public void run() {
            // 心跳检测逻辑
        }
    };

    checkHeartbeatThread = new Thread(check, "tio-timer-checkheartbeat-" + id + "-" + name);
    checkHeartbeatThread.setDaemon(true);
    checkHeartbeatThread.setPriority(Thread.MIN_PRIORITY);
    checkHeartbeatThread.start();
}

  3. 心跳检测线程

    心跳检测线程定期扫描所有连接，判断是否超时未响应，进行必要的处理。

    java
    Runnable check = new Runnable() {
    @Override
    public void run() {
        // 初始等待
        Thread.sleep(1000 * 10);

        while (needCheckHeartbeat && !isStopped()) {
            if (heartbeatTimeout <= 0) {
                break;
            }
            try {
                Thread.sleep(heartbeatTimeout);
            } catch (InterruptedException e1) {
                log.error(e1.toString(), e1);
            }
            long start = SystemTimer.currTime;
            SetWithLock<ChannelContext> setWithLock = connections;
            Set<ChannelContext> set = null;
            ReadLock readLock = setWithLock.readLock();
            readLock.lock();
            try {
                set = setWithLock.getObj();

                for (ChannelContext channelContext : set) {
                    long compareTime = Math.max(channelContext.stat.latestTimeOfReceivedByte, channelContext.stat.latestTimeOfSentPacket);
                    long currtime = SystemTimer.currTime;
                    long interval = currtime - compareTime;

                    boolean needRemove = false;
                    if (channelContext.heartbeatTimeout != null && channelContext.heartbeatTimeout > 0) {
                        needRemove = interval > channelContext.heartbeatTimeout;
                    } else {
                        needRemove = interval > heartbeatTimeout;
                    }

                    if (needRemove) {
                        if (!ServerTioConfig.this.serverAioListener.onHeartbeatTimeout(channelContext, interval, channelContext.stat.heartbeatTimeoutCount.incrementAndGet())) {
                            log.info("{}, {} ms or not send and receive message", channelContext, interval);
                            channelContext.setCloseCode(CloseCode.HEARTBEAT_TIMEOUT);
                            Tio.remove(channelContext, interval + " ms not send and receive message");
                        }
                    }
                }
            } catch (Throwable e) {
                log.error("", e);
            } finally {
                try {
                    readLock.unlock();
                    if (debug) {
                        diagnostic(start, set, start1, count);
                    }
                } catch (Throwable e) {
                    log.error("", e);
                }
            }
        }
    }
};

技术细节解析 ​

  1. 心跳机制

    心跳机制通过定期检查连接的最近接收和发送时间，判断连接是否活跃。如果超过超时时间未有数据交互，则触发超时处理，包括调用监听器的onHeartbeatTimeout方法，并可能关闭连接。

  2. 线程与并发

    • 使用ReentrantReadWriteLock管理连接集合的并发访问，确保高效的读写操作。
    • 心跳检测运行在一个独立的守护线程中，确保不会阻塞主线程。

  3. 统计与日志

    • 统计信息包括已接受连接数、当前连接数、发送和接收的数据量等。
    • 在调试模式下，定期输出详细的诊断信息，帮助开发者监控系统状态。

服务器启动与监听：TioServer ​

TioServer类负责启动服务器，绑定端口，监听连接，并处理接收到的连接请求。

类结构与成员变量 ​

java
public class TioServer {
    private static Logger log = LoggerFactory.getLogger(TioServer.class);
    private ServerTioConfig serverTioConfig;
    private AsynchronousServerSocketChannel serverSocketChannel;
    private Node serverNode;
    private boolean isWaitingStop = false;
    private boolean checkLastVersion = true;
    private ExecutorService groupExecutor;
    private AsynchronousChannelGroup channelGroup;

    // 构造函数与其他方法略
}

主要成员变量解析 ​

  1. 配置与节点

    • serverTioConfig：服务器配置对象。
    • serverNode：服务器节点信息，包括 IP 和端口。

  2. 网络通信

    • serverSocketChannel：异步服务器套接字通道，负责接受客户端连接。
    • channelGroup：异步通道组，管理一组异步通道。

  3. 状态管理

    • isWaitingStop：标志服务器是否正在等待停止。
    • checkLastVersion：版本检查开关。

  4. 线程与执行器

    • groupExecutor：用于管理异步通道的线程池。

关键方法解析 ​

  1. 启动服务器

    start()方法负责初始化服务器配置，绑定端口，启动监听，并接受连接。

    java
    public void start(String serverIp, int serverPort) throws IOException {
    serverTioConfig.init();
    serverTioConfig.getCacheFactory().register(TioCoreConfigKeys.REQEUST_PROCESSING, null, null, null);

    this.serverNode = new Node(serverIp, serverPort);
    if (EnvUtils.getBoolean("tio.core.hotswap.reload", false)) {
        groupExecutor = Threads.getGroupExecutor();
        channelGroup = AsynchronousChannelGroup.withThreadPool(groupExecutor);
        serverSocketChannel = AsynchronousServerSocketChannel.open(channelGroup);
    } else {
        serverSocketChannel = AsynchronousServerSocketChannel.open();
    }

    serverSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
    serverSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 64 * 1024);

    InetSocketAddress listenAddress = null;

    if (StrUtil.isBlank(serverIp)) {
        listenAddress = new InetSocketAddress(serverPort);
    } else {
        listenAddress = new InetSocketAddress(serverIp, serverPort);
    }

    serverSocketChannel.bind(listenAddress, 0);

    AcceptCompletionHandler acceptCompletionHandler = new AcceptCompletionHandler();
    serverSocketChannel.accept(this, acceptCompletionHandler);
    serverTioConfig.startTime = System.currentTimeMillis();
    Threads.getTioExecutor();
}

    步骤解析：

    • 初始化配置：调用serverTioConfig.init()，确保所有配置和状态被正确初始化。
    • 绑定端口：根据提供的 IP 和端口，绑定服务器套接字通道。
    • 启动监听：创建AcceptCompletionHandler实例，开始接受客户端连接。
    • 心跳线程：初始化和启动心跳检测线程，确保连接的活跃性。

  2. 停止服务器

    stop()方法负责优雅地关闭服务器，释放资源，并关闭所有连接。

    java
    public boolean stop() {
    isWaitingStop = true;

    if (channelGroup != null) {
        try {
            channelGroup.shutdownNow();
        } catch (Exception e) {
            log.error("Faild to execute channelGroup.shutdownNow()", e);
        }
    }

    if (groupExecutor != null) {
        try {
            groupExecutor.shutdownNow();
        } catch (Exception e) {
            log.error("Failed to close groupExecutor", e);
        }

    }

    if (serverSocketChannel != null) {
        try {
            serverSocketChannel.close();
        } catch (Exception e) {
            log.error("Failed to close serverSocketChannel", e);
        }

    }

    serverTioConfig.setStopped(true);
    boolean ret = Threads.close();
    log.info(this.serverNode + " stopped");
    return ret;
}

    步骤解析：

    • 标记停止：设置isWaitingStop为true，通知其他组件服务器即将停止。
    • 关闭异步通道组：尝试关闭channelGroup，终止所有关联的异步操作。
    • 关闭执行器：关闭线程池groupExecutor，释放线程资源。
    • 关闭服务器套接字通道：关闭serverSocketChannel，停止接受新连接。
    • 更新配置状态：设置serverTioConfig的停止标志，并关闭线程池。

  3. 版本检查

    checkLastVersion用于控制是否进行版本检查，但在提供的代码中仅记录了日志，未实际实现版本检查逻辑。

    java
    public void setCheckLastVersion(boolean checkLastVersion) {
    log.debug("community edition is no longer supported");
}

连接接收处理：AcceptCompletionHandler ​

AcceptCompletionHandler 实现 CompletionHandler<AsynchronousSocketChannel, TioServer>。成功回调先检查停服与监听通道状态，再独立提交下一次 accept，随后初始化当前连接。

初始化与异常清理 ​

初始化过程包括获取客户端地址、检查 IP 黑名单、设置 socket 选项、建立 ServerChannelContext、通知连接监听器、分配初始 VirtualBuffer，以及提交首次读取。

首次 read 成功提交之前，接收回调负责持有的 socket、上下文和缓冲区。任何提前退出或初始化异常都会进入清理流程；上下文已登记时通过 Tio.close 清理，最后关闭原始 socket 兜底。首次 read 成功提交之后，缓冲区交给读取回调。

以下代码节选展示交接点，省略连接配置与监听器通知：

java
ReadCompletionHandler readCompletionHandler = new ReadCompletionHandler(channelContext);
attachment = BufferPoolUtils.allocateRequest(channelContext.getReadBufferSize());
ByteBuffer readByteBuffer = attachment.buffer();
readByteBuffer.position(0);
readByteBuffer.limit(readByteBuffer.capacity());
clientSocketChannel.read(readByteBuffer, attachment, readCompletionHandler);
attachment = null;
handedOff = true;

重新监听使用独立的异常边界。检查服务仍在运行且监听通道打开后调用 accept；失败时记录 Failed to rearm accept; check the listening channel state。这能保住当前已接收连接，但后续接收能力仍需检查。

接收失败与回调失败 ​

底层 accept 失败会在重置 pending 状态后通知本次 failed。completed 回调自身抛出的异常则单独处理：记录异常并关闭本次 socket，不能再通过可变 handler 字段调用 failed，否则可能误伤回调中刚提交的下一次 accept。

Worker 将单次操作的异常处理与连接资源管理相结合，支持共享 IO 线程持续处理后续事件。设计说明见 t-io 稳定性设计与资源管理。

数据读取与解码：ReadCompletionHandler 与 DecodeTask ​

ReadCompletionHandler 当前实现 CompletionHandler<Integer, VirtualBuffer>。VirtualBuffer 是异步回调的 attachment，buffer() 返回其中的 ByteBuffer。旧示例直接使用 ByteBuffer 作为 attachment，无法反映当前缓冲池的释放责任。

读取、解码与下一次读取 ​

正数读取结果先更新连接统计，再把 ByteBuffer 切换到读模式。普通连接交给 DecodeTask 解码；SSL 连接先复制密文并交给 SSL facade 解密。解密后的明文经 SslListener.onPlainData 交给 SslFacadeContext 持有的解码器，继续完成半包处理与业务分发。

完成数据处理后，只有 TioUtils.checkBeforeIO(channelContext) 通过才会继续读取。读取缓冲区大小发生变化时，先释放旧 VirtualBuffer，再申请新的缓冲区。

以下是下一次读取的交接点：

java
if (byteBuffer.capacity() != channelContext.getReadBufferSize()) {
  release(current);
  current = null;
  current = BufferPoolUtils.allocateRequest(channelContext.getReadBufferSize());
  byteBuffer = current.buffer();
}
byteBuffer.clear();
channelContext.asynchronousSocketChannel.read(byteBuffer, current, this);
submitted = true;

当前回调用 submitted 记录是否已把缓冲区交给下一次 read。finally 只在未提交成功时释放 current；不能无条件释放仍在被异步读取使用的缓冲区。分配或提交失败时，释放当前仍持有的缓冲区并清理连接。

终止和失败分支 ​

  • 结果为零、EOF 或其他负值时，按相应关闭原因清理连接并释放缓冲区。
  • 解码异常使用 DECODE_ERROR，SSL 解密异常使用 SSL_DECRYPT_ERROR；其他意外错误使用 READ_ERROR。
  • failed 对所有失败类型执行清理，不仅限于 ClosedChannelException。
  • Tio.close 清理抛出异常时，记录错误并尝试直接关闭 socket。

failed 的资源释放结构如下：

java
@Override
public void failed(Throwable exc, VirtualBuffer virtualBuffer) {
  try {
    closeSafely(exc, "Failed to read data", ChannelCloseCode.READ_ERROR);
  } finally {
    release(virtualBuffer);
  }
}

IO 日志按处理阶段区分位置；操作结果与 attachment 的管理方式见 t-io 稳定性设计与资源管理。

消息处理：HandlePacketTask ​

HandlePacketTask类负责将解码后的业务包交给业务逻辑处理器，进行具体的业务处理。

类结构与成员变量 ​

java
public class HandlePacketTask {
    private AtomicLong synFailCount = new AtomicLong();

    public void handler(ChannelContext channelContext, Packet packet) {
        // 处理业务包的逻辑
    }
}

关键方法解析 ​

  1. 业务包处理

    handler()方法负责将业务包传递给 AIO 处理器，并更新统计信息。

    java
    public void handler(ChannelContext channelContext, Packet packet) {
    TioConfig tioConfig = channelContext.tioConfig;

    if (packet.isKeepConnection() && !channelContext.isBind) {
        tioConfig.ips.bind(channelContext);
        tioConfig.ids.bind(channelContext);
        channelContext.groups = new SetWithLock<String>(new HashSet<>());
        channelContext.isBind = true;
    }
    long start = SystemTimer.currTime;
    try {
        Integer synSeq = packet.getSynSeq();
        if (synSeq != null && synSeq > 0) {
            MapWithLock<Integer, Packet> syns = tioConfig.getWaitingResps();
            Packet initPacket = syns.remove(synSeq);
            if (initPacket != null) {
                synchronized (initPacket) {
                    syns.put(synSeq, packet);
                    initPacket.notify();
                }
            } else {
                log.error("[{}] Failed to synchronize message, synSeq is {}, but there is no corresponding key value in the synchronization collection", synFailCount.incrementAndGet(), synSeq);
            }
        } else {
            if (EnvUtils.getBoolean(TioCoreConfigKeys.TIO_CORE_DIAGNOSTIC, false)) {
                Long id = packet.getId();
                String requestInfo = channelContext.getClientIpAndPort() + "_" + id;
                log.info("handle:{}", requestInfo);
            }
            tioConfig.getAioHandler().handler(packet, channelContext);
        }
    } catch (Throwable e) {
        e.printStackTrace();
    } finally {
        long end = SystemTimer.currTime;
        long iv = end - start;
        if (tioConfig.statOn) {
            channelContext.stat.handledPackets.incrementAndGet();
            channelContext.stat.handledBytes.addAndGet(packet.getByteCount());
            channelContext.stat.handledPacketCosts.addAndGet(iv);

            tioConfig.groupStat.handledPackets.incrementAndGet();
            tioConfig.groupStat.handledBytes.addAndGet(packet.getByteCount());
            tioConfig.groupStat.handledPacketCosts.addAndGet(iv);
        }

        if (CollUtil.isNotEmpty(tioConfig.ipStats.durationList)) {
            try {
                for (Long v : tioConfig.ipStats.durationList) {
                    IpStat ipStat = (IpStat) tioConfig.ipStats.get(v, channelContext);
                    ipStat.getHandledPackets().incrementAndGet();
                    ipStat.getHandledBytes().addAndGet(packet.getByteCount());
                    ipStat.getHandledPacketCosts().addAndGet(iv);
                    tioConfig.getIpStatListener().onAfterHandled(channelContext, packet, ipStat, iv);
                }
            } catch (Exception e1) {
                e1.printStackTrace();
            }
        }

        if (tioConfig.getAioListener() != null) {
            try {
                tioConfig.getAioListener().onAfterHandled(channelContext, packet, iv);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

    步骤解析：

    • 连接绑定：如果业务包要求保持连接（isKeepConnection），且连接尚未绑定，则将连接绑定到 IP、ID 等。

    • 同步请求处理：

      • 如果业务包带有同步序列号（synSeq），则从等待响应的集合中查找对应的请求包，并通知等待的线程。
      • 如果未找到对应的请求包，则记录同步失败计数，并记录错误日志。

    • 异步请求处理：

      • 如果业务包没有同步序列号，则直接调用 AIO 处理器的handler()方法，进行业务处理。

    • 统计信息更新：记录处理包的数量、字节数和处理时间。

    • 回调监听器：通知 AIO 监听器业务包已处理完毕。

技术细节解析 ​

  1. 连接绑定

    对于需要保持连接的业务包，通过绑定 IP 和 ID，确保连接的状态与业务逻辑的一致性。

  2. 同步与异步请求处理

    • 同步请求：通过序列号匹配请求与响应，实现同步通信的效果。使用CountDownLatch等待响应。
    • 异步请求：直接将业务包交给 AIO 处理器，无需等待，适用于高并发场景。

  3. 统计与日志

    • 通过AtomicLong和SetWithLock管理统计信息，确保高效的并发处理。
    • 在调试模式下，记录详细的处理日志，便于问题排查。

  4. 异常处理

    • 捕获并记录业务处理过程中的异常，确保系统的稳定性。
    • 对于同步请求未能找到对应的请求包，记录同步失败计数，预防潜在的同步问题。

网络通信管理：Tio ​

Tio类作为 Tio 框架的核心 API，提供了丰富的方法用于管理连接、发送消息、关闭连接等操作。以下将详细解析其关键功能和技术实现。

类结构与成员变量 ​

java
public class Tio {
    public static class IpBlacklist {
        // IP黑名单相关方法
    }

    private static Logger log = LoggerFactory.getLogger(Tio.class);

    // 其他内部类和方法略

    private Tio() {}
}

主要功能解析 ​

  1. IP 黑名单管理

    IpBlacklist内部类提供了对 IP 黑名单的管理功能，包括添加、删除、检查等。

    java
    public static class IpBlacklist {
    public static boolean add(TioConfig tioConfig, String ip) {
        return tioConfig.ipBlacklist.add(ip);
    }

    public static boolean add(String ip) {
        return GlobalIpBlacklist.INSTANCE.global.add(ip);
    }

    public static void clear(TioConfig tioConfig) {
        tioConfig.ipBlacklist.clear();
    }

    public static void clear() {
        GlobalIpBlacklist.INSTANCE.global.clear();
    }

    public static Collection<String> getAll(TioConfig tioConfig) {
        return tioConfig.ipBlacklist.getAll();
    }

    public static Collection<String> getAll() {
        return GlobalIpBlacklist.INSTANCE.global.getAll();
    }

    public static boolean isInBlacklist(TioConfig tioConfig, String ip) {
        if (tioConfig.ipBlacklist != null) {
            return tioConfig.ipBlacklist.isInBlacklist(ip) || GlobalIpBlacklist.INSTANCE.global.isInBlacklist(ip);
        } else {
            return GlobalIpBlacklist.INSTANCE.global.isInBlacklist(ip);
        }
    }

    public static void remove(TioConfig tioConfig, String ip) {
        tioConfig.ipBlacklist.remove(ip);
    }

    public static void remove(String ip) {
        GlobalIpBlacklist.INSTANCE.global.remove(ip);
    }
}

    技术细节解析：

    • 全局与局部黑名单：支持针对特定TioConfig的局部黑名单和全局黑名单，增强灵活性。
    • 并发访问：通过SetWithLock和其他并发集合管理黑名单，确保线程安全。

  2. 发送消息

    Tio类提供了多种发送消息的方法，包括发送到单个连接、群组、指定 IP 等。

    java
    public static Boolean send(ChannelContext channelContext, Packet packet) {
    return send(channelContext, packet, null, null);
}

private static Boolean send(ChannelContext channelContext, Packet packet, CountDownLatch countDownLatch, PacketSendMode packetSendMode) {
    // 发送逻辑
}

public static Boolean bSend(ChannelContext channelContext, Packet packet) {
    if (channelContext == null) {
        return false;
    }
    CountDownLatch countDownLatch = new CountDownLatch(1);
    return send(channelContext, packet, countDownLatch, PacketSendMode.SINGLE_BLOCK);
}

// 其他发送方法略

    关键方法解析：

    • send()：核心发送方法，负责将消息包传递给SendPacketTask进行发送，并处理同步发送的逻辑。

    • bSend()：阻塞发送方法，适用于需要同步等待发送结果的场景。通过CountDownLatch实现同步等待。

    • 群组与指定 IP 发送：通过遍历目标群组或 IP 对应的连接集合，批量发送消息，提高系统的广播能力。

  3. 关闭连接

    Tio提供了多种关闭连接的方法，包括关闭单个连接、群组、指定 IP 等。

    java
    public static void close(ChannelContext channelContext, String remark) {
    close(channelContext, null, remark);
}

public static void close(ChannelContext channelContext, Throwable throwable, String remark) {
    close(channelContext, throwable, remark, false);
}

// 其他关闭方法略

    关键方法解析：

    • close()：核心关闭方法，负责关闭连接并执行必要的清理操作。

    • remove()：与close()方法类似，但不执行重连等维护操作，适用于需要立即移除连接的场景。

  4. 群组管理

    Tio提供了方法用于绑定和解绑群组，管理连接的群组关系。

    java
    public static void bindGroup(ChannelContext channelContext, String group) {
    channelContext.tioConfig.groups.bind(group, channelContext);
}

public static void unbindGroup(String group, ChannelContext channelContext) {
    channelContext.tioConfig.groups.unbind(group, channelContext);
}

// 其他群组管理方法略

    技术细节解析：

    • 高效的群组管理：通过SetWithLock和并发集合管理群组内的连接，确保高效的群组操作。

    • 灵活的群组操作：支持动态的群组绑定与解绑，适应复杂的业务需求。

技术细节解析 ​

  1. 并发与锁机制

    • 使用ReentrantLock和Condition管理异步操作的同步，确保多线程环境下的操作安全。
    • SetWithLock和MapWithLock封装了线程安全的集合操作，简化了并发编程。

  2. 心跳与连接检测

    • 通过心跳检测机制，定期检查连接的活跃性，及时关闭失效连接，提升系统的稳定性。
    • 使用AtomicLong和原子变量管理统计信息，确保高效的并发更新。

  3. SSL 支持

    • 通过SslConfig和SslFacadeContext实现 SSL 加密，确保数据传输的安全性。
    • 在发送和接收数据时，自动进行加密与解密，简化开发者的操作。

  4. 缓冲区管理

    • 使用ByteBufferPool管理缓冲区，减少内存分配与回收的开销，提升系统性能。
    • 通过ByteBufferUtils提供的工具方法，简化缓冲区操作。

  5. 错误处理与日志

    • 通过CompletionHandler接口的failed()方法统一处理读取与写入过程中的异常，确保系统的健壮性。
    • 在不同的错误场景下，记录详细的日志信息，便于开发者排查问题。

异常处理与连接关闭：CloseTask ​

CloseTask类负责管理连接的关闭过程，包括释放资源、更新状态、处理重连等。

类结构与成员变量 ​

java
public class CloseTask {
    public static void close(ChannelContext channelContext) {
        // 关闭连接的逻辑
    }
}

关键方法解析 ​

  1. 关闭连接

    close()方法负责关闭连接，并根据配置决定是否进行重连或清理操作。

    java
    public static void close(ChannelContext channelContext) {
    boolean isNeedRemove = channelContext.closeMeta.isNeedRemove;
    String remark = channelContext.closeMeta.remark;
    Throwable throwable = channelContext.closeMeta.throwable;
    channelContext.stat.timeClosed = SystemTimer.currTime;
    try {
        if (channelContext.tioConfig.getAioListener() != null) {
            try {
                channelContext.tioConfig.getAioListener().onBeforeClose(channelContext, throwable, remark, isNeedRemove);
            } catch (Throwable e) {
                log.error(e.toString(), e);
            }
        }

        try {
            if (channelContext.isClosed && !isNeedRemove) {
                return;
            }

            if (channelContext.isRemoved) {
                return;
            }

            try {
                if (isNeedRemove) {
                    MaintainUtils.remove(channelContext);
                } else {
                    ClientTioConfig clientTioConfig = (ClientTioConfig) channelContext.tioConfig;
                    clientTioConfig.closeds.add(channelContext);
                    clientTioConfig.connecteds.remove(channelContext);
                    MaintainUtils.close(channelContext);
                }

                channelContext.setRemoved(isNeedRemove);
                if (channelContext.tioConfig.statOn) {
                    channelContext.tioConfig.groupStat.closed.incrementAndGet();
                }
                channelContext.stat.timeClosed = SystemTimer.currTime;
                channelContext.setClosed(true);
            } catch (Throwable e) {
                log.error(e.toString(), e);
            } finally {
                if (!isNeedRemove && channelContext.isClosed && !channelContext.isServer()) {
                    ClientChannelContext clientChannelContext = (ClientChannelContext) channelContext;
                    ReconnConf.put(clientChannelContext);
                }
            }
        } catch (Throwable e) {
            log.error(throwable.toString(), e);
        }
    } finally {
        channelContext.isWaitingClose = false;
    }
}

    步骤解析：

    • 状态检查：判断连接是否已经关闭或被移除，避免重复操作。

    • 调用监听器：如果配置了 AIO 监听器，调用onBeforeClose()方法，通知监听器连接即将关闭。

    • 资源释放：

      • 移除连接：如果需要移除，则从群组、用户等管理集合中移除连接。
      • 更新统计：更新全局和连接级别的关闭统计信息。
      • 设置状态：标记连接为已关闭和已移除。

    • 重连处理：对于客户端配置，如果连接被关闭且不需要移除，则将连接加入重连队列，尝试重新连接。

    • 异常处理：捕获并记录关闭过程中的异常，确保系统稳定性。

技术细节解析 ​

  1. 资源管理

    在关闭连接时，确保所有相关资源（如缓冲区、引用等）被正确释放，防止内存泄漏和资源浪费。

  2. 状态同步

    通过设置标志位（如isClosed、isRemoved）确保连接状态的一致性，避免并发环境下的状态混乱。

  3. 重连机制

    对于客户端连接，在连接被关闭后，根据重连配置决定是否尝试重新连接，增强系统的容错性和可用性。

  4. 监听器回调

    提供钩子机制，通过 AIO 监听器的回调方法，让开发者在连接关闭前执行自定义逻辑，如资源清理、日志记录等。

  5. 多线程支持

    在多线程环境下，通过使用ReentrantLock和Condition，确保关闭操作的线程安全性。

示例运行与日志分析 ​

基于用户提供的日志，以下是一个请求的处理流程解析。

测试请求 ​

bash
curl http://localhost:8000/ok

处理显示的日志 ​

2024-12-06 18:23:40.134 [Thread-2] INFO  c.l.t.s.AcceptCompletionHandler.completed:68 - new connection:127.0.0.1,42972
2024-12-06 18:23:40.134 [Thread-2] INFO  c.l.t.c.t.DecodeTask.decode:41 - decode:127.0.0.1:42972
2024-12-06 18:23:40.134 [Thread-2] INFO  c.l.t.c.t.HandlePacketTask.handler:62 - handle:127.0.0.1:42972_281494
2024-12-06 18:23:40.135 [Thread-1] INFO  c.l.t.c.ReadCompletionHandler.completed:104 - close 127.0.0.1:42972, because The connection closed by peer
2024-12-06 18:23:40.135 [Thread-1512] INFO  c.l.t.c.WriteCompletionHandler.completed:57 - write 172 to 127.0.0.1:42972
2024-12-06 18:23:40.135 [Thread-1] INFO  c.l.t.c.Tio.close:455 - close 127.0.0.1:42972,remark:The connection closed by peer

日志解析 ​

  1. 连接建立

    2024-12-06 18:23:40.134 [Thread-2] INFO  c.l.t.s.AcceptCompletionHandler.completed:68 - new connection:127.0.0.1,42972
    • 事件：新连接建立。
    • 信息：客户端 IP 为127.0.0.1，端口为42972。
    • 处理类：AcceptCompletionHandler的completed()方法。

  2. 数据解码

    2024-12-06 18:23:40.134 [Thread-2] INFO  c.l.t.c.t.DecodeTask.decode:41 - decode:127.0.0.1:42972
    • 事件：接收到的数据开始解码。
    • 信息：来自127.0.0.1:42972的数据正在解码。
    • 处理类：DecodeTask的decode()方法。

  3. 业务包处理

    2024-12-06 18:23:40.134 [Thread-2] INFO  c.l.t.c.t.HandlePacketTask.handler:62 - handle:127.0.0.1:42972_281494
    • 事件：业务包开始处理。
    • 信息：处理来自127.0.0.1:42972的业务包，包 ID 为281494。
    • 处理类：HandlePacketTask的handler()方法。

  4. 读取完成，连接关闭

    2024-12-06 18:23:40.135 [Thread-1] INFO  c.l.t.c.ReadCompletionHandler.completed:104 - close 127.0.0.1:42972, because The connection closed by peer
    • 事件：连接读取完成，且对端关闭连接。
    • 信息：因为对端关闭连接，127.0.0.1:42972的连接被关闭。
    • 处理类：ReadCompletionHandler的completed()方法。

  5. 写入完成

    2024-12-06 18:23:40.135 [Thread-1512] INFO  c.l.t.c.WriteCompletionHandler.completed:57 - write 172 to 127.0.0.1:42972
    • 事件：数据写入完成。
    • 信息：写入172字节到127.0.0.1:42972。
    • 处理类：WriteCompletionHandler的completed()方法。

  6. 连接关闭处理

    2024-12-06 18:23:40.135 [Thread-1] INFO  c.l.t.c.Tio.close:455 - close 127.0.0.1:42972,remark:The connection closed by peer
    • 事件：连接正式关闭。
    • 信息：关闭127.0.0.1:42972的连接，备注信息为The connection closed by peer。
    • 处理类：Tio类的close()方法。

技术细节解析 ​

  1. 连接生命周期

    • 建立：通过AcceptCompletionHandler接受新连接，创建ChannelContext，并开始异步读取数据。
    • 数据处理：读取到数据后，通过ReadCompletionHandler进行解码，并交由HandlePacketTask处理业务逻辑。
    • 连接关闭：对端关闭连接后，ReadCompletionHandler触发关闭流程，调用Tio.close()进行连接的清理和资源释放。

  2. 异步操作与回调

    • 非阻塞通信：使用 Java AIO 的异步通道，实现高效的非阻塞网络通信。
    • 回调机制：通过实现CompletionHandler接口，处理异步操作完成后的回调，实现事件驱动的通信模型。

  3. 错误处理

    • 读取异常：在读取过程中，如果出现异常，调用Tio.close()关闭连接，确保系统的稳定性。
    • 写入异常：在写入过程中，如果出现异常，同样调用Tio.close()关闭连接。

  4. 统计与监控

    • 连接统计：统计已接收连接数、当前连接数、发送和接收的字节数等，提供系统的运行状态监控。
    • 心跳检测：通过心跳机制，检测连接的活跃性，及时关闭失效连接。

总结与技术要点 ​

本文基于提供的代码片段，详细解析了 Tio 框架的核心配置、服务器启动与监听、连接接收处理、数据读取与解码、消息处理、网络通信管理、异常处理与连接关闭等关键组件和功能。通过深入理解这些组件的结构和交互方式，开发者可以更好地应用 Tio 框架，构建高性能、稳定的网络应用。

关键技术要点 ​

  1. 高效的异步通信：基于 Java AIO 的异步通道，实现高并发下的高效网络通信。
  2. 灵活的配置管理：通过TioConfig和ServerTioConfig，提供丰富的配置选项，满足不同场景需求。
  3. 心跳与连接检测：通过心跳机制和统计信息，确保连接的活跃性和系统的稳定性。
  4. IP 黑名单与安全：支持 IP 黑名单功能，增强系统的安全性，防止恶意连接。
  5. 消息编码与解码：通过 AIO 处理器和解码任务，灵活处理不同协议和业务逻辑的数据包。
  6. 群组与用户管理：提供群组、用户、令牌等多种连接管理方式，适应复杂的业务需求。
  7. 异常处理与日志记录：统一的异常处理机制和详细的日志记录，帮助开发者快速定位和解决问题。
  8. 资源管理与性能优化：通过缓冲区池和并发集合，优化资源管理和系统性能，减少内存开销。

通过深入理解 Tio 框架的这些关键技术要点，开发者可以充分发挥其优势，构建高性能、稳定、安全的网络应用。

参考文献 ​

tiocloud 文档资料

MIT Licensed

Copyright © 2023-present litongjava