在 Skynet 中，Actor 是基本的并发单元，其调度由一个线程池负责。每个 Actor 都可以被视为一个独立的、拥有自身状态和消息队列的对象。Skynet 通过将所有处于活跃状态的 Actor 消息队列串联成一个全局链表（或队列）来实现调度。线程池中的工作线程会从这个全局队列中取出活跃的 Actor 消息队列进行消息处理。

最初的设计理念是实现“公平调度”，即每个活跃的 Actor 都能获得处理消息的机会。然而，由于不同 Actor 的消息队列长度可能差异巨大，如果工作线程持续处理一个拥有大量消息的 Actor，可能会导致其他消息量较少或消息处理速度较慢的 Actor 出现“饥饿”现象，即长时间无法获得 CPU 时间来处理其消息。

为了规避这种“饥饿”问题，Skynet 引入了

机制。它通过赋予线程池中的工作线程不同的权重来调节它们每次从特定 Actor 消息队列中处理消息的数量。具体来说，一个工作线程在处理某个 Actor 的消息时，会根据自身的权重，决定从该 Actor 的消息队列中取出多少条消息进行处理。权重越高，该工作线程在一次调度周期内从单个 Actor 队列中处理的消息数量越少，从而使得该 Actor 的消息队列有机会更快地被放回全局队列，让其他工作线程或在后续调度中处理其他 Actor，以此确保所有活跃 Actor 都能得到公平的调度机会，有效防止了“饥饿”现象的发生。

一个 Actor 在 Skynet 中可以被抽象为：

一个隔离的运行环境 + 一个回调函数 + 一个专属消息队列

Actor 的调度最终由工作线程执行，因此整个调度流程的起点是线程池中的工作线程入口函数 

static void *
thread_worker(void *p) {
        struct worker_parm *wp = p;
        int id = wp->id;
        int weight = wp->weight; // 工作线程的权重
        struct monitor *m = wp->m;
        struct skynet_monitor *sm = m->m[id];
        skynet_initthread(THREAD_WORKER); // 初始化线程环境
        struct message_queue * q = NULL; // 当前处理的Actor消息队列
        while (!m->quit) { // 主循环，直到Skynet退出
                // 核心调度逻辑：从全局队列中获取并处理Actor消息
                q = skynet_context_message_dispatch(sm, q, weight);
                if (q == NULL) { // 如果全局队列为空，没有活跃的Actor消息可处理
                        if (pthread_mutex_lock(&m->mutex) == 0) {
                                ++ m->sleep; // 标记线程进入睡眠状态
                                // "虚假唤醒"无害，因为skynet_context_message_dispatch()可以随时调用。
                                if (!m->quit)
                                        pthread_cond_wait(&m->cond, &m->mutex); // 线程等待条件变量，进入休眠
                                -- m->sleep; // 线程被唤醒，退出睡眠状态
                                if (pthread_mutex_unlock(&m->mutex)) {
                                        fprintf(stderr, "unlock mutex error");
                                        exit(1);
                                }
                        }
                }
        }
        return NULL;
}

 函数作为消费者，在一个无限循环中不断尝试从全局消息队列中获取并处理 Actor 消息。如果当前没有可处理的消息（即 

 进入休眠状态，等待新的消息到来或被唤醒。

 是实际负责取出消息并分发给对应 Actor 处理的核心函数。

 结构体保存了每个 Actor 的上下文信息，是 Actor 运行状态的封装。它定义在 

struct skynet_context {
        void * instance;        // 指向Actor的实例数据，通常是Lua虚拟机（lua_State*）
        struct skynet_module * mod; // Actor所属的模块
        void * cb_ud;           // 回调函数的自定义数据
        skynet_cb cb;           // Actor的消息处理回调函数（C函数）
        struct message_queue *queue; // Actor专属的消息队列
        ATOM_POINTER logfile;   // 日志文件指针
        uint64_t cpu_cost;      // CPU耗时（微秒），用于性能分析
        uint64_t cpu_start;     // CPU计时开始时间
        char result[32];        // 结果缓冲区
        uint32_t handle;        // Actor的唯一标识ID（句柄）
        int session_id;         // 会话ID
        ATOM_INT ref;           // 引用计数
        int message_count;      // 已处理消息计数
        bool init;              // 是否已初始化
        bool endless;           // 是否永不停止
        bool profile;           // 是否开启性能分析

        CHECKCALLING_DECL       // 调试宏
};

 通常指向一个 Lua 虚拟机实例，代表了 Actor 的运行环境；

 是 Actor 注册的消息处理回调函数，当消息到达时，Skynet 会调用此函数来驱动 Actor 运行；

 则是该 Actor 独有的消息队列，所有发往此 Actor 的消息都将存入此队列。

每个 Actor 还拥有一个专属的消息队列 

struct message_queue {
        struct spinlock lock;   // 保护消息队列的自旋锁
        uint32_t handle;        // 消息队列所属Actor的句柄
        int cap;                // 队列容量
        int head;               // 队列头部索引
        int tail;               // 队列尾部索引
        int release;            // 释放标志
        int in_global;          // 标记该队列是否在全局消息队列中
        int overload;           // 队列过载计数
        int overload_threshold; // 队列过载阈值
        struct skynet_message *queue; // 存储消息的数组
        struct message_queue *next; // 指向全局消息队列中的下一个消息队列
};

该结构体包含了消息队列的锁、容量、头尾指针以及是否在全局队列中的状态等信息。

 函数是调度流程的核心，它负责从全局队列中取出活跃的 Actor 消息队列，并根据工作线程的权重处理其中的消息。该函数定义在 

struct message_queue *
skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q, int weight) {
        if (q == NULL) { // 如果没有传入待处理的队列（第一次调度或上一个队列已处理完毕）
                q = skynet_globalmq_pop(); // 从全局活跃消息队列中弹出一个Actor的消息队列
                if (q==NULL) // 如果全局队列为空，表示没有活跃的Actor
                        return NULL;
        }

        uint32_t handle = skynet_mq_handle(q); // 获取当前消息队列所属Actor的句柄

        struct skynet_context * ctx = skynet_handle_grab(handle); // 根据句柄获取Actor的上下文
        if (ctx == NULL) { // 如果Actor上下文不存在（可能已被销毁）
                struct drop_t d = { handle };
                skynet_mq_release(q, drop_message, &d); // 释放该消息队列中的消息
                return skynet_globalmq_pop(); // 继续尝试获取下一个Actor队列
        }

        int i,n=1; // n代表本次调度周期内将处理的消息数量
        struct skynet_message msg;

        // 循环处理消息
        for (i=0;i<n;i++) {
                if (skynet_mq_pop(q,&msg)) { // 从Actor的专属消息队列中弹出一个消息
                        skynet_context_release(ctx); // 释放Actor上下文引用
                        return skynet_globalmq_pop(); // 如果队列为空，则尝试获取下一个Actor队列
                } else if (i==0 && weight >= 0) { // 第一次处理消息时，根据权重计算本次处理的消息数量
                        n = skynet_mq_length(q); // 获取当前Actor消息队列的长度
                        n >>= weight; // 根据工作线程的权重调整处理消息的数量，n = n / (2^weight)
                }
                int overload = skynet_mq_overload(q); // 检查队列是否过载
                if (overload) {
                        skynet_error(ctx, "May overload, message queue length = %d", overload);
                }

                skynet_monitor_trigger(sm, msg.source , handle); // 触发监控

                if (ctx->cb == NULL) { // 如果Actor没有设置回调函数
                        skynet_free(msg.data); // 直接释放消息数据
                } else {
                        dispatch_message(ctx, &msg); // 调用dispatch_message分发消息给Actor的回调函数
                }

                skynet_monitor_trigger(sm, 0,0); // 触发监控

        }

        assert(q == ctx->queue); // 确保处理的是Actor自己的消息队列
        struct message_queue *nq = skynet_globalmq_pop(); // 尝试从全局队列中获取下一个Actor队列
        if (nq) {
                // 如果全局队列不为空，将当前处理完部分消息的队列重新放回全局队列，并返回下一个队列
                // 否则（全局队列为空或阻塞），不放回当前队列，并返回当前队列（下次继续处理）
                skynet_globalmq_push(q); // 将当前Actor的消息队列重新放回全局活跃队列
                q = nq; // 更新为下一个待处理的Actor队列
        }
        skynet_context_release(ctx); // 释放Actor上下文引用

        return q; // 返回下一个待处理的Actor队列，或当前队列（如果全局队列为空）
}

 函数是消息处理的最终环节，它负责调用 Actor 注册的 C 回调函数来实际处理接收到的消息。该函数定义在 

static void
dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
        assert(ctx->init); // 确保Actor已初始化
        CHECKCALLING_BEGIN(ctx) // 调试宏，检查调用栈
        // 设置线程私有数据，使当前线程知道正在处理哪个Actor的消息
        pthread_setspecific(G_NODE.handle_key, (void *)(uintptr_t)(ctx->handle));
        int type = msg->sz >> MESSAGE_TYPE_SHIFT; // 提取消息类型
        size_t sz = msg->sz & MESSAGE_TYPE_MASK; // 提取消息大小
        FILE *f = (FILE *)ATOM_LOAD(&ctx->logfile); // 获取日志文件指针
        if (f) {
                skynet_log_output(f, msg->source, type, msg->session, msg->data, sz); // 写入日志
        }
        ++ctx->message_count; // 增加消息计数
        int reserve_msg;
        if (ctx->profile) { // 如果开启性能分析
                ctx->cpu_start = skynet_thread_time(); // 记录CPU开始时间
                // 调用Actor的C回调函数处理消息
                reserve_msg = ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz);
                uint64_t cost_time = skynet_thread_time() - ctx->cpu_start; // 计算CPU耗时
                ctx->cpu_cost += cost_time; // 累加CPU总耗时
        } else {
                // 调用Actor的C回调函数处理消息（无性能分析）
                reserve_msg = ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz);
        }
        if (!reserve_msg) { // 如果回调函数没有保留消息数据
                skynet_free(msg->data); // 释放消息数据
        }
        CHECKCALLING_END(ctx) // 调试宏
}

三、C 语言到 Lua 的调用过程分析

Skynet 的 Actor 大部分业务逻辑是用 Lua 编写的，但底层的调度和消息分发是在 C 语言层面完成的。因此，需要一个机制将 C 语言层面的消息分发最终桥接到 Lua 层的 Actor 业务逻辑。

这个桥接是通过 C 语言向 Lua 导出 

 中，它的作用是让 Lua 代码能够注册一个回调函数，这个回调函数最终会被 C 语言层面的 

static int
lcallback(lua_State *L) {
        // 从Lua的upvalue中获取skynet_context指针
        struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
        int forward = lua_toboolean(L, 2); // 获取是否为转发模式

        luaL_checktype(L,1,LUA_TFUNCTION); // 检查第一个参数是否为函数
        lua_settop(L,1); // 保持栈顶只有回调函数

        // 创建一个callback_context用户数据，用于保存Lua回调函数的上下文
        struct callback_context *cb_ctx = (struct callback_context *)lua_newuserdata(L, sizeof(*cb_ctx));
        cb_ctx->L = lua_newthread(L); // 为回调函数创建一个新的Lua协程（thread）
        lua_pushcfunction(cb_ctx->L, traceback); // 将traceback函数压入新协程的栈顶，用于错误追踪
        lua_setuservalue(L, -2); // 将traceback函数作为用户值关联到cb_ctx用户数据
        lua_setfield(L, LUA_REGISTRYINDEX, "callback_context"); // 将cb_ctx存储到Lua注册表中

        lua_xmove(L, cb_ctx->L, 1); // 将lcallback的第一个参数（Lua函数）移动到新创建的协程中

        if (forward) {
                skynet_callback(context, cb_ctx, forward_cb); // 注册转发模式的回调
        } else {
                skynet_callback(context, cb_ctx, _cb); // 注册普通模式的回调
        }

        return 0;
}

 模块中，使得 Lua 代码可以通过 

LUAMOD_API int
luaopen_skynet_core(lua_State *L) {
        // ... (其他函数注册)

        luaL_Reg l[] = {
                // ...
                { "callback", lcallback }, // 注册lcallback函数
                // ...
        };

        // ...
        luaL_setfuncs(L,l,1); // 将l数组中的函数注册到Lua表中，lcallback会捕获skynet_context作为upvalue
        // ...
}

 来设置 Actor 的消息处理入口。

-- /lualib/skynet.lua
function skynet.start(start_func)
        c.callback(skynet.dispatch_message) -- 注册skynet.dispatch_message作为C层回调
        init_thread = skynet.timeout(0, function()
                skynet.init_service(start_func)
                init_thread = nil
        end)
end

 函数被调用以处理一个新消息时，它会通过 

 函数会负责唤醒并执行之前在 

 中创建的 Lua 协程，最终调用到 Lua 层的 

 函数来处理具体的消息。这样就实现了从 C 语言底层调度到 Lua 业务逻辑的无缝衔接。

本文深入分析了 Skynet 框架中的 Actor 调度机制。首先，阐述了其基于线程池的公平调度原理，并重点解释了如何通过工作线程的“权重”机制来有效解决 Actor 消息队列长度不一致导致的“饥饿”问题。

 等核心 C 语言源码的详细解析，揭示了 Skynet 调度循环、消息分发以及 Actor 上下文管理的内部工作流程。

最后，剖析了 C 语言与 Lua 之间消息处理回调的桥接过程，特别是 

 函数的作用，展现了 Skynet 如何实现跨语言的 Actor 消息处理。通过这些分析，读者可以更全面地理解 Skynet 框架高效、健壮的并发调度设计。

源码分析Skynet的Actor对等调度：理解不一样的任务调度机制

Skynet 是一款轻量级并发框架，其核心在于 Actor 模型的高效调度。以下是对 Skynet Actor 调度机制的详细阐述，包括其公平调度策略、源码实现解析以及 C 语言与 Lua 之间的调用过程。

架构设计

后端

开发工具

Skynet框架采用线程池实现Actor公平调度，通过权重机制防饥饿。解析thread_worker等核心源码，揭示消息分发流程，介绍C与Lua间回调桥接，展现高效并发设计。

源码分析Skynet的Actor对等调度：理解不一样的任务调度机制

源码分析Skynet的Actor对等调度：理解不一样的任务调度机制

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐