skynet源码学习-clusteragent.lua
- 一、文件概述
- 二、模块依赖和初始化
- 2.1 模块引入
- 2.2 启动参数处理
- 三、全局变量和状态管理
- 3.1 large_request - 大请求分片缓存
- 3.2 inquery_name - 名称查询等待队列
- 3.3 register_name_mt - 名称缓存元表
- 3.4 new_register_name 函数
- 3.5 tracetag 变量
- 四、核心请求分发函数
- 4.1 函数名和参数
- 4.2 初始处理
- 4.3 分片请求处理
- 4.4 请求类型处理
- 4.5 响应打包和发送
- 五、协议注册和服务初始化
- 5.1 注册客户端协议
- 5.2 转发网关数据
- 5.3 Lua 消息处理
- 六、关键设计原理
- 6.1 连接模型
- 6.2 分片处理机制
- 6.3 名称服务设计
- 6.4 跟踪系统集成
- 6.5 错误处理策略
- 七、性能优化设计
- 7.1 零拷贝优化
- 7.2 批量写入
- 7.3 并发控制
- 7.4 缓存策略
- 八、与其他模块的交互
- 8.1 与 clusterd 的关系
- 8.2 与 gate 的关系
- 8.3 与 clustersender 的关系
- 8.4 与本地服务的关系
- 九、使用场景和示例
- 9.1 正常请求流程
- 9.2 名称查询流程
- 9.3 大消息处理流程
- 十、限制和注意事项
- 10.1 连接限制
- 10.2 内存管理
- 10.3 性能考虑
这是 Skynet 集群模块中的入站连接处理器，负责处理来自其他节点的连接请求，并将请求转发给本地服务。是 Skynet集群能够实现高性能、可靠的跨节点通信的关键组件之一
一、文件概述
clusteragent.lua 是 Skynet 集群模块中的入站连接处理器，它：
- 处理来自其他节点的连接请求
- 解析集群协议并转发给本地服务
- 管理服务名称查询和缓存
- 处理大消息分片和重组
- 支持跟踪（trace）功能传递
核心角色
- 服务器端处理器：与 clustersender（客户端）对应
- 协议解码器：解析集群协议格式
- 本地服务路由器：将请求路由到正确的本地服务
- 名称服务缓存：缓存远程节点查询的名称
服务启动参数
localclusterd,gate,fd=...clusterd=tonumber(clusterd)gate=tonumber(gate)fd=tonumber(fd)- clusterd: 集群核心服务地址
- gate: 网关服务地址（接收网络连接）
- fd: 连接文件描述符（网络套接字）
二、模块依赖和初始化
localskynet=require"skynet"localsocket=require"skynet.socket"localcluster=require"skynet.cluster.core"localignoreret=skynet.ignoreretlocalclusterd,gate,fd=...clusterd=tonumber(clusterd)gate=tonumber(gate)fd=tonumber(fd)详细解析
2.1 模块引入
- skynet: Skynet 核心库
- skynet.socket: Socket 操作模块
- 提供 socket.write, socket.lwrite 等函数
- skynet.cluster.core: 集群核心协议模块
- 提供 cluster.unpackrequest, cluster.packresponse 等函数
- skynet.ignoreret: 特殊函数
- 作用: 忽略返回值，避免自动回复
- 在特定场景下使用（如 session 是 fd 时）
2.2 启动参数处理
clusterd=tonumber(clusterd)gate=tonumber(gate)fd=tonumber(fd)- 类型转换: 所有参数都是字符串，需要转换为数字
- 参数来源: 由 clusterd 服务在创建代理时传入
- 典型调用:
-- 在 clusterd.lua 的 socket 处理中localagent=skynet.newservice("clusteragent",skynet.self(),-- clusterdsource,-- gatefd-- 连接文件描述符)三、全局变量和状态管理
locallarge_request={}localinquery_name={}localregister_namelocalregister_name_mt={__index=function(self,name)localwaitco=inquery_name[name]ifwaitcothenlocalco=coroutine.running()table.insert(waitco,co)skynet.wait(co)returnrawget(register_name,name)elsewaitco={}inquery_name[name]=waitcolocaladdr=skynet.call(clusterd,"lua","queryname",name:sub(2))-- name must be '@xxxx'ifaddrthenregister_name[name]=addrendinquery_name[name]=nilfor_,coinipairs(waitco)doskynet.wakeup(co)endreturnaddrendend}localfunctionnew_register_name()register_name=setmetatable({},register_name_mt)endnew_register_name()localtracetag详细解析
3.1 large_request - 大请求分片缓存
locallarge_request={}- 结构: {session = 分片请求表}
- 作用: 存储正在接收的分片请求，直到所有分片到达
- 键: 请求的会话ID（session）
- 值: 分片请求表，包含：
{addr=目标地址,is_push=是否推送,tracetag=跟踪标签,-- 分片数据通过 cluster.append 添加}3.2 inquery_name - 名称查询等待队列
localinquery_name={}- 结构: {名称 = 等待协程数组}
- 作用: 管理并发的名称查询，避免重复查询
- 场景: 多个请求同时查询同一个名称时
- 设计思想: 第一个查询进行实际查询，后续查询等待结果
3.3 register_name_mt - 名称缓存元表
这是一个关键设计，实现了名称查询的延迟加载和并发控制。
元表定义
localregister_name_mt={__index=function(self,name)-- 查询逻辑end}查询函数详细解析
第1步：检查是否已有查询在进行
localwaitco=inquery_name[name]ifwaitcothenlocalco=coroutine.running()table.insert(waitco,co)skynet.wait(co)returnrawget(register_name,name)end场景：已有其他协程正在查询同一名称
- 获取等待队列: local waitco = inquery_name[name]
- 加入队列: table.insert(waitco, co)
- co: 当前协程
- 挂起等待: skynet.wait(co)
- 等待查询完成
- 返回结果: return rawget(register_name, name)
- 直接获取缓存结果
- 注意: 使用 rawget 避免再次触发元表
第2步：开始新的查询
waitco={}inquery_name[name]=waitcolocaladdr=skynet.call(clusterd,"lua","queryname",name:sub(2))-- name must be '@xxxx'ifaddrthenregister_name[name]=addrendinquery_name[name]=nilfor_,coinipairs(waitco)doskynet.wakeup(co)endreturnaddr详细流程：
- 创建等待队列: waitco = {}
- 标记为查询中: inquery_name[name] = waitco
- 发起实际查询:
localaddr=skynet.call(clusterd,"lua","queryname",name
