skynet源码学习-clusteragent.lua

这是 Skynet 集群模块中的入站连接处理器，负责处理来自其他节点的连接请求，并将请求转发给本地服务。是 Skynet集群能够实现高性能、可靠的跨节点通信的关键组件之一

一、文件概述

clusteragent.lua 是 Skynet 集群模块中的入站连接处理器，它：

1. 处理来自其他节点的连接请求
2. 解析集群协议并转发给本地服务
3. 管理服务名称查询和缓存
4. 处理大消息分片和重组
5. 支持跟踪（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 模块引入

1. skynet: Skynet 核心库
2. skynet.socket: Socket 操作模块

• 提供 socket.write, socket.lwrite 等函数

3. skynet.cluster.core: 集群核心协议模块

• 提供 cluster.unpackrequest, cluster.packresponse 等函数

4. 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

场景：已有其他协程正在查询同一名称

1. 获取等待队列: local waitco = inquery_name[name]
2. 加入队列: table.insert(waitco, co)

• co: 当前协程

3. 挂起等待: skynet.wait(co)

• 等待查询完成

4. 返回结果: 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

详细流程：

1. 创建等待队列: waitco = {}
2. 标记为查询中: inquery_name[name] = waitco
3. 发起实际查询:

localaddr=skynet.call(clusterd,"lua","queryname",name