UnLua中实现基于RPC的网络数据收发机制

byteyang

概述

这篇文档说明一套跑在 UnLua 上的网络收发机制。业务仍然走 UE 的 RPC，协议变化时只增加 proto 和消息号，RPC 函数保持不变。所有业务数据都封成 protobuf 字节，由少数几个固定 RPC 负责运送。消息号、回复配对、监听和发送队列放在 Lua 里。

只要加一条新消息，读到「业务侧怎么用」就可以动手。后面的「底层怎么串起来」说明这些 API 为什么长这样，以及队列、回复、可靠与不可靠各自管什么。

为什么需要这一层

UE 的 RPC 是一次远程函数调用，参数类型在 C++ 或蓝图里声明。数据一复杂，就要先定义结构体，再定义函数。在 C++ 和蓝图里这还能接受，类型是现成的。

放到 Lua 里会多出一整段反射工作：先在蓝图声明 RPC，复杂结构再建一份 USTRUCT，然后在 Lua 里创建这个结构体、填字段、再调用 RPC。每加一条协议都要改蓝图或 C++，协议一多，声明和业务逻辑就散在两处。

这里用游戏里常见的做法。协议结构写在 proto 里，运行时用 protobuf 编成一段字节。RPC 只保留固定的几条，参数是消息号加上这段字节。业务侧看到的是消息号和一张 Lua table。编解码、监听、请求队列都收在 Network 管理器里。

整体结构

每个玩家的 PlayerController 上挂一个 URPCComponent。组件通过 UnLua 绑到 RPCComponent.lua。业务通过 _G.Client.Network 或 _G.DS.Network 收发。管理器负责查消息名、做 protobuf 编解码、维护监听和等待队列，最后通过这个组件把字节送出去。

业务、管理器、组件、UE RPC 的关系如下。左右两侧各有一套 Lua，中间只有固定的那几条 RPC 穿过网络。

组件上只有四条 RPC，覆盖这个机制里的全部流量：

RPC

方向

可靠性

携带的内容

做什么

C2S_ReqMessage

客户端 → DS

可靠

消息号、回复号、字节

客户端发起一条消息

S2C_RspMessage

DS → 客户端

可靠

消息号、回复号、错误码、字节

DS 回复前面某一次请求

S2C_NtfMessage

DS → 客户端

可靠

消息号、字节

DS 主动推送

S2C_NtfMessage_UnReliable

DS → 客户端

不可靠

消息号、字节

DS 主动推送，允许丢失

客户端发往 DS 的请求始终走可靠 RPC。不可靠只用于 DS 下行的推送，适合高频、丢一包也可以靠下一次状态覆盖的数据。

一条需要回复的请求按这个顺序走。队列里若还有下一条，回调执行完会接着把它发出去。

不需要回复的请求走另一条路。RspUID 传 0，编码后立即发送，进入等待队列的只有带回调的请求。DS 收到后只分发给监听，不为它占等待槽。

DS 主动告诉客户端某件事时，同样走推送通道。SendMessage 发给某一个玩家，BroadcastMessage 发给当前所有玩家。可靠与否由参数决定，默认可靠。广播时字节只编码一次，再分别塞进每个玩家的组件。

消息要先约定四样东西

运行时按消息号查表，把一个数字还原成一段 protobuf。这依赖四份约定。前三份把消息号还原成 protobuf，错误码给回复回调用。

结构写在 proto 里。 新建 proto，使用固定的 package。管理器拼类型名时用的就是这个 package 加上消息名。拼出来的类型名形如 com.tencent.wea.protocol.xxx.Demo_Req，package 和消息名之间是一个点。前缀不一致时，编码和解码会找不到类型。

syntax = "proto2";
package com.tencent.wea.protocol.xxx;

message Demo_Req {
  optional int32 Id = 1;
}

message Demo_Rsp {
  optional int32 Id = 1;
}

message Demo_Ntf {
  optional int32 Id = 1;
}

消息号写在 MessageEnum.lua。 业务收发时只认这个数字。E_NONE = 0 表示回复里没有消息体，真实消息从非 0 的值开始分配。

_G.MessageEnum = {
    E_NONE = 0,
    E_DEMO_REQ = 1101,
    E_DEMO_RSP = 1102,
    E_DEMO_NTF = 1103,
}

消息号和 proto 消息名的对应关系写在 MessageName.lua。 管理器用这张表取出 Demo_Req 这种名字，再和 package 拼成 pb 的类型名。表里的字符串要和 proto 里的 message 名一致。

local MessageEnum = _G.MessageEnum
_G.MessageName = {
    [MessageEnum.E_DEMO_REQ] = "Demo_Req",
    [MessageEnum.E_DEMO_RSP] = "Demo_Rsp",
    [MessageEnum.E_DEMO_NTF] = "Demo_Ntf",
}

错误码写在 MessageError.lua。 E_SUCCESS = 0 表示成功。回复回调的第一个参数就是这个错误码。0 是合法值。判断「调用方有没有传入错误码」时要和 nil 比，if not nErrorCode 会把成功当成没有传入。

_G.MessageError = {
    E_SUCCESS = 0,
}

业务侧怎么用

客户端发给 DS

SendMessage 的三个参数是消息号、消息 table、回复回调。回调可以不传。

传了回调，就表示这次请求要等 DS 回复。请求进入串行队列：同一时刻只有队首在路上，这一次等待结束之后，才会发送下一条。回调的第一个参数是错误码，第二个参数是回复 table。DS 只回了错误码时，第二个参数为空。

不传回调，请求立即发送，等待队列不记录它。DS 侧对应的处理函数里，回复号是 0。

_G.Client.Network:SendMessage(_G.MessageEnum.E_DEMO_REQ, tMessage, function(nErrorCode, tRspMessage)
    Log("GM_DemoRsp", nErrorCode, tRspMessage and tRspMessage.Id)
end)

客户端接收 DS 的推送

推送由 DS 主动发出，和客户端那条请求队列无关。谁关心某条消息，谁就按消息号注册。同一个消息号可以有多个监听对象，每个对象对同一个消息号只能注册一次。回调的参数是解码后的 table。

对象销毁或界面关闭时要注销，管理器按对象挂着这些回调。可以按对象一次卸掉它注册过的全部消息，也可以只卸某一个消息号。

_G.Client.Network:RegisterMessage(_G.MessageEnum.E_DEMO_NTF, self, self.On_DemoNtf)

function XXX:On_DemoNtf(tMessage)
    Log("GM_DemoNtf", tMessage.Id)
end

_G.Client.Network:UnRegisterWithScope(self)
_G.Client.Network:UnRegisterMessage(_G.MessageEnum.E_DEMO_NTF, self)

DS 发给客户端

指定玩家用 SendMessage，所有玩家用 BroadcastMessage。最后一个参数表示是否走不可靠通道，不传则走可靠推送。不可靠推送可以丢失，也可以晚于后发的可靠消息到达。业务要能接受「这一包没有到」，也要能接受「收到的是更早的状态」。

_G.DS.Network:SendMessage(nPlayerUID, _G.MessageEnum.E_DEMO_NTF, tNtfMessage, bUnReliable)
_G.DS.Network:BroadcastMessage(_G.MessageEnum.E_DEMO_NTF, tNtfMessage, bUnReliable)

DS 处理客户端请求

DS 上每个消息号只能注册一个处理函数。同一个消息号再注册一次会被拒绝，避免两处业务先后回复同一次请求。

处理函数的参数依次是：玩家 UID、解码后的消息 table、回复函数 fReply、这次请求的回复号。

玩家带了回复回调时，nRspUID 大于 0，处理函数负责调用 fReply。可以带回一个消息体，也可以只回错误码。回复体用哪个消息号，由处理函数传入，一般是成对的 Rsp 消息。

回复可以晚于处理函数返回。例如要等玩法的异步结果时，把 nPlayerUID 和 nRspUID 存下来，之后调用 ReplyMessage。客户端只认队首上的那个回复号，回复号丢了，回包会被丢弃。

玩家没带回调时，nRspUID 是 0。这种请求没有人在等，fReply 没有接收方。

_G.DS.Network:RegisterMessage(_G.MessageEnum.E_DEMO_REQ, self, self.On_DemoReq)

function XXX:On_DemoReq(nPlayerUID, tMessage, fReply, nRspUID)
    local tRspMessage = { Id = tMessage.Id + 100 }
    fReply(_G.MessageError.E_SUCCESS, _G.MessageEnum.E_DEMO_RSP, tRspMessage)
end

-- 只回错误码：
-- fReply(_G.MessageError.E_SUCCESS)
-- 异步完成后再回复。同一次请求不要既调用 fReply，又调用 ReplyMessage。
-- _G.DS.Network:ReplyMessage(nPlayerUID, nRspUID, nErrorCode, _G.MessageEnum.E_DEMO_RSP, tRspMessage)

DS 注销时，按消息号卸掉的是这个消息唯一的处理函数；按对象卸掉的是这个对象注册过的全部消息。

_G.DS.Network:UnRegisterMessage(nMessageId)
_G.DS.Network:UnRegisterWithScope(self)

底层怎么串起来

字节从 Lua 到 RPC

URPCComponent 挂在 PlayerController 上，每个玩家一条通道。DS 要发给谁，就找谁的 Controller，再拿到上面的组件。客户端只拿自己的那一个。

Lua 里 protobuf 编出来的是一段可以包含 \0 的二进制。UE 反射若用 FString 接这段数据，会在第一个 \0 处截断。C++ 和 Lua 之间按长度交接字节：Lua 侧是 lua_pushlstring / lua_tolstring，UE 侧是 TArray<uint8>。RPC 参数仍然是普通的 int32 加 TArray<uint8>，UE 的复制通道保持原样。

一条消息在两端之间的形态变化如下。去程和回程用的是同一套转换，方向相反。

以客户端发请求为例，交接是两次方向相反的转换。

Lua 调用导出的 C2S_ReqMessageWithBytes。导出函数从栈上取出消息号、回复号和二进制串，装进 TArray<uint8>，再调用 C2S_ReqMessage。这条函数标记为 Server, Reliable，由 UE 送到 DS。

DS 的 _Implementation 做相反的事：把 TArray<uint8> 拷成一块连续内存，用 lua_pushlstring 按长度压栈，然后调用组件上的 C2S_ReqMessageWithBytes_Lua。Lua 函数再交给 DS.Network:ReceiveMessage。

下行的三条 RPC 是同一套转换。差别只在参数：回复多一个错误码；不可靠推送的 UFUNCTION 标记是 Client, Unreliable。进入 Lua 之后，可靠推送和不可靠推送都走 NtfMessage，业务回调看不到它来自哪一条 RPC。

真正越过 Lua 和 C++ 边界的，是下面这两组函数。管理器不直接调用 UFUNCTION。Lua 先调导出的 C 函数，C 函数把二进制装进 TArray<uint8> 再调用 RPC。对端的 _Implementation 收到数组后，按长度压回 Lua，调用组件上对应的 *_Lua 函数。

Lua 到 TArray 按长度用 memcpy 拷贝，字节里可以含有 \0。RPC 交回 Lua 时直接把数组数据指针交给 lua_pushlstring，空数组压空串。

static void BytesToArray(const char* Bytes, size_t Size, TArray<uint8>& Array)
{
    if (Bytes == nullptr || Size == 0)
    {
        return;
    }

    const int32 Offset = Array.Num();
    Array.AddUninitialized(static_cast<int32>(Size));
    FMemory::Memcpy(Array.GetData() + Offset, Bytes, Size);
}

组件上只声明四条 RPC。消息号、回复号、错误码都是 int32，载荷是 TArray<uint8>。

UFUNCTION(Server, Reliable)
void C2S_ReqMessage(int32 MessageId, int32 RspUID, const TArray<uint8>& Message);

UFUNCTION(Client, Reliable)
void S2C_RspMessage(int32 MessageId, int32 RspUID, int32 ErrorCode, const TArray<uint8>& Message);

UFUNCTION(Client, Reliable)
void S2C_NtfMessage(int32 MessageId, const TArray<uint8>& Message);

UFUNCTION(Client, Unreliable)
void S2C_NtfMessage_UnReliable(int32 MessageId, const TArray<uint8>& Message);

从 Lua 发出 RPC

Lua 里写 RPCComponent:C2S_ReqMessageWithBytes(MessageId, RspUID, bytes) 时，实际进的是导出函数，不是 UFUNCTION 本身。导出函数用 lua_tolstring 取出指针和长度。空载荷允许是 nil，这时发出去的数组是空的。

static int32 URPCComponent_C2S_ReqMessageWithBytes(lua_State* L)
{
    if (lua_gettop(L) < 3)
        return luaL_error(L, "invalid parameters");

    URPCComponent* Self = Get(L, 1, UnLua::TType<URPCComponent*>());
    if (!IsValid(Self))
        return luaL_error(L, "self is invalid");

    const int32 MessageId = lua_tointeger(L, 2);
    const int32 RspUID = lua_tointeger(L, 3);

    TArray<uint8> Message;
    if (lua_isnil(L, 4) == false)
    {
        size_t Size = 0;
        const char* Bytes = lua_tolstring(L, 4, &Size);
        BytesToArray(Bytes, Size, Message);
    }

    Self->C2S_ReqMessage(MessageId, RspUID, Message);
    return 0;
}

下行的两个导出函数是同一套读栈。S2C_RspMessageWithBytes 的顺序是 self、消息号、回复号、错误码、字节，再调用 S2C_RspMessage。S2C_NtfMessageWithBytes 的顺序是 self、消息号、是否不可靠、字节，第三个参数为真时调用 S2C_NtfMessage_UnReliable，否则调用 S2C_NtfMessage。可靠和不可靠在进 RPC 之前就分开，Lua 业务不自己选 UFUNCTION。

Get 来自 UnLua 头文件。组件需要 UCLASS，导出宏才能挂上这些函数。三个导出函数注册到组件上，Lua 才能当方法调用：

static constexpr luaL_Reg URPCComponentLib[] =
{
    {"C2S_ReqMessageWithBytes", URPCComponent_C2S_ReqMessageWithBytes},
    {"S2C_RspMessageWithBytes", URPCComponent_S2C_RspMessageWithBytes},
    {"S2C_NtfMessageWithBytes", URPCComponent_S2C_NtfMessageWithBytes},
    {nullptr, nullptr},
};

BEGIN_EXPORT_REFLECTED_CLASS(URPCComponent)
    ADD_LIB(URPCComponentLib)
END_EXPORT_CLASS()
IMPLEMENT_EXPORTED_CLASS(URPCComponent)

从 RPC 交回 Lua

UE 把 RPC 送到对端后，进的是 *_Implementation。这里不用 UnLua 的反射去调 Lua。反射如果走 FString，二进制会在第一个 \0 被截断。所以直接拿 lua_State，用 lua_pushlstring 按 TArray 的长度压栈。

以 DS 收到客户端请求为例。栈先记下当前高度，压入错误处理函数，再从组件对象上取出 C2S_ReqMessageWithBytes_Lua。lua_getfield 不会弹出对象，函数压在对象上面。调用时要再压一次对象，作为 Lua 函数的 self。后面三个参数是消息号、回复号和字节。lua_pcall 的参数个数是 4，错误处理函数的栈位置是进入函数时记下的 Top + 1。调用结束后把栈恢复到进入时的高度。lua_pcall 失败时错误处理函数会打日志，这条 RPC 仍然结束，对端要靠超时才知道没有业务回调。

void URPCComponent::C2S_ReqMessage_Implementation(int32 MessageId, int32 RspUID, const TArray<uint8>& Message)
{
    lua_State* L = UnLua::GetState();
    if (!L)
    {
        return;
    }

    const int32 Top = lua_gettop(L);
    lua_pushcfunction(L, UnLua::ReportLuaCallError);
    UnLua::PushUObject(L, this);
    if (lua_getfield(L, -1, "C2S_ReqMessageWithBytes_Lua") == LUA_TFUNCTION)
    {
        UnLua::PushUObject(L, this);
        lua_pushinteger(L, MessageId);
        lua_pushinteger(L, RspUID);
        if (Message.Num() > 0)
        {
            lua_pushlstring(L, reinterpret_cast<const char*>(Message.GetData()), static_cast<size_t>(Message.Num()));
        }
        else
        {
            lua_pushlstring(L, "", 0);
        }
        lua_pcall(L, 4, 0, Top + 1);
    }
    lua_settop(L, Top);
}

另外三条 _Implementation 只改压栈的参数和要调用的 Lua 函数名：

RPC

调用的 Lua 函数

压栈参数

S2C_RspMessage

S2C_RspMessageWithBytes_Lua

self、消息号、回复号、错误码、字节，共 5 个

S2C_NtfMessage

S2C_NtfMessageWithBytes_Lua

self、消息号、字节，共 3 个

S2C_NtfMessage_UnReliable

S2C_NtfMessageWithBytes_UnReliable_Lua

self、消息号、字节，共 3 个

不可靠推送进 Lua 之后，仍然交给 Client.Network:NtfMessage。业务回调分不出这一包是从可靠 RPC 还是不可靠 RPC 进来的。

Lua 侧只做转交，解释字节的工作在管理器里：

function RPCComponent:C2S_ReqMessageWithBytes_Lua(MessageID, RspUID, Message)
    _G.DS.Network:ReceiveMessage(self.UID, MessageID, RspUID, Message)
end

function RPCComponent:S2C_RspMessageWithBytes_Lua(MessageID, RspUID, ErrorCode, Message)
    _G.Client.Network:RspMessage(MessageID, RspUID, ErrorCode, Message)
end

function RPCComponent:S2C_NtfMessageWithBytes_Lua(MessageID, Message)
    _G.Client.Network:NtfMessage(MessageID, Message)
end

function RPCComponent:S2C_NtfMessageWithBytes_UnReliable_Lua(MessageID, Message)
    _G.Client.Network:NtfMessage(MessageID, Message)
end

玩家 UID 取组件上的 self.UID。C++ 不把 UID 当参数传入。这个字段在 DS 上由游戏写入，客户端消息体里的字段不会改变它。

管理器负责协议

发送前，管理器用 MessageName 取出消息名，和固定的 package 前缀拼成 pb 类型名，例如 com.tencent.wea.protocol.xxx.Demo_Req。这个字符串按消息号缓存，避免每次发送都拼接。可靠 RPC 超出 UE 通道的单包上限时，发送在引擎层失败，Lua 收不到错误回调。然后 pb.encode 把 table 编成字节，交给组件。

接收时顺序相反：先确认消息号有对应的名字，再 pb.decode，最后查监听表并调用回调。回调用 xpcall 包起来，一个业务函数出错时，同一次推送里的其余监听仍会被调用。

参数里的消息号、回复号、错误码都可以是 0。判断「调用方有没有传这个值」时写成 == nil。if not nRspUID 会把「不需要回复」的请求直接丢掉，if not nErrorCode 会把成功回复直接丢掉。

客户端的等待队列

只有带回调的发送才进队列。队列元素记下回复号、消息号、已编码的字节、回调和入队时间。

入队后如果自己就是队首，立刻调用组件发送。前面还有未完成的请求时，这次只入队。队首的回包到来，弹出队首之后再发下一条。

回包到达时，只承认当前队首，并且回复号必须一致。对上之后执行回调。回复号对不上的包直接丢掉，迟到的回包就不会配到后面那条请求上。

队首是否等得太久，在下一次 SendMessage 时检查，而且发生在当前请求入队之前。示例里的阈值是 10 秒。旧队首超时后，先清空当时已经在队列里的请求，对其中每个回调传入超时错误码，再把当前这条入队并发送。当前这条不在被清空的队列里。队列自己不计时。之后再也没有新的发送时，队首会一直留在队列里。

回复号从 1 递增，在 1 到 2^31-1 之间回绕。0 保留给不需要回复的发送。C++ 侧这个参数是 int32，超过这个范围会被截断，截断后可能变成 0，和「不需要回复」冲突。

找不到 RPCComponent 时，SendMessage 直接返回，没有返回值。这次请求不进队列。带回调的调用收不到回复，也不会等到超时，只能从日志里看到这次没有送出去。

DS 的等待槽

DS 保存每个玩家当前那一个待回复请求，保存的是槽位而不是客户端那条队列。ReceiveMessage 看到 RspUID > 0 时，如果该玩家还没有槽，就记下回复号和时间。槽还在，说明上一条需要回复的请求还没结束，这次仍然分发给业务，同时打出重复请求的日志。

处理函数拿到的 fReply 会调用 ReplyMessage，把错误码和可选消息体送回这个回复号，然后清掉该玩家的槽。槽按玩家保存，不按回复号保存。后一次 fReply 清掉的是当前槽，槽里记的可能已经不是这次的回复号。RspUID == 0 时不占槽。

管理器的 Tick 大约每秒看一次这些槽。示例里超过 3 秒就丢掉槽位，让这个玩家的下一次请求能够再占上。Tick 只释放槽，业务侧的回复仍然要自己发。处理函数如果在这之后才调用 fReply，回复仍按原来的回复号发出。客户端队首还是同一个回复号时，回调仍能对上。

监听表为什么两边不一样

客户端同一条推送经常要通知多个界面。一个消息号下面是「对象 → 回调」的一张表。另外用「对象 → 消息号」的反表，UnRegisterWithScope 就能在对象销毁时一次卸完。

DS 上同一条请求只有一处业务负责处理和回复。一个消息号只保存一个处理函数。重复注册直接拒绝。

使用时要记住的行为

  • 带回复回调的客户端请求是串行的。前一条没有回包，后一条停在队列里。高频、不需要结果的消息不要传回调。

  • RspUID 为 0 表示这条消息没有回复。错误码 0 表示成功。两处都是合法数值。

  • 客户端的超时检查挂在下一次发送上，示例阈值 10 秒。超时会让当时已经在队列里的回调都收到超时，当前这条新请求不在其中。DS 的超时检查在 Tick 里，示例阈值 3 秒，只释放该玩家的等待槽。

  • 异步回复要带上原来的 nRspUID。客户端只接受队首上的那个回复号。

  • 不可靠推送可以丢失，也可以晚于后发的可靠推送到达。

  • 找不到 RPCComponent 时，这次发送直接返回，没有返回值，也不会进入等待队列。业务收不到回复，也不会等到超时。

评论