这是最后一篇了，其实还有很多重要的模块要分析的，但留给以后有多余时间再去研究吧，有兴趣的可以自行下载源码分析。这部分主要是围绕第三小问题展开，并附加些其他skynet中与此有关的设计，即：当并发时，如何保证消息的正确时序，以及如何使用协程处理消息(同步/异步/超时)；包括创建协程处理消息，挂起协程，切换。这块其实是针对lua上层来说的，底层框架的消息队列只是保证消息顺序入队列且出队列，如果交叉执行比如lua层的协程挂起，那么就会出现时序问题。

先简单回顾下前几篇博客的分析，包括skynet本身的设计，及C++协程。对于C++协程，比如一个请求a过来后，从协程池中pop一个协程并处理该请求a，如果需要等待，则让出协程并挂上定时器，然后再处理下一个请求b，如果此时a和b是相关联的，且b有可能依赖于a的执行结果，那么就会出现问题。这对于游戏中的业务来说，尤其涉及到金钱相关的逻辑，那是大问题。而那种独立的请求间，只是读之类的操作，那是没问题的。如果需要结合业务，那么就需要改造。

而对于skynet来说，当并发上来时，考虑到这个时序问题，底层实现相关的顺序队列，大概思路就是lua协程执行a到一半后，哪怕有b的消息被协程调度处理，此时会把这个b协程压入队列(lua中的table也可以，使用数组部分)，必须等a执行完毕或超时后，再处理b的，也就是在业务上层串行化了服务的消息处理。这样保证了时序。

但这又引起了另一个问题，即可能存在后面的消息都超时了，然而上层如果无法识别继续处理，那么就白白浪费了资源，处理了无用的消息。这类的相关介绍在另一篇“谈谈缓存穿透雪崩和过载保护以及一致性等问题”中有相关的介绍及应对方案。

本节分为两个小点讨论，即：
1)如何保证消息的正确时序，以及如何使用协程处理消息(同步/异步/超时)；
2)创建协程处理消息，挂起协程，切换；

第一小点，撇开语言方面的限制，考虑skynet本身的框架设计，而不掺杂业务框架的设计。对于单进程多线程，要想并发的处理同一个客户端的请求，不管是读还是写，都必须路由到同一个线程处理，这样就保证了不会导致同一个client的请求分发到不同的线程，在skynet底层抽象client为一个agent service，有自己的消息队列，并且当工作线程处理这个agent消息时，先把这个消息队列从全局队列出摘出来，从这个队列pop一条消息，处理完毕后，再把这个消息队列挂到全局队列中；而对于push消息到agent队列则没有这种过程，只要获得自旋锁即可，相关源码可以见前面的分析。

这一层就保证了消息不会乱序，但是对于业务层，使用lua协程来提高并发，那么就要好好设计。

这里举例比如在主场景中，这样可以考虑到client的所有消息都路由到场景后需要考虑到的时序问题。
当与client有关的两条有依赖关系的消息a和b被场景服务dispatch分发处理时，不考虑读还是写，都会创建一个协程，并执行相关的处理函数。比如数据安全性不是特别严重的例子，玩家在帮派中，然后点领取今日奖励b消息，此时帮主把玩家踢出帮派a消息，本来是a先执行完毕后再b执行的顺序，这时可能出现a先执行导致挂起，而b执行完毕后，接着执行a的情况，多领了一份奖励。当然这里只是为了举例，通过检查可以避免这种问题。

简单分析下，在skynet的做法中，为每个服务加个lua层的消息队列，进入该队列的消息会被依次处理完毕，不管中间是否挂起，这样带来的问题是，并发度降底了且引入了一定的复杂度。

 17     dispatch = function(session, from, ...)
 18         table.insert(message_queue, {session = session, addr = from, ... })
 19         if thread_id then //有消息，如果有等待则wakeup
 20             skynet.wakeup(thread_id)
 21             thread_id = nil
 22         end
 23     end

 26 local function do_func(f, msg)
 27     return pcall(f, table.unpack(msg))
 28 end
 29 
 30 local function message_dispatch(f)
 31     while true do
 32         if #message_queue==0 then  //没消息则挂起
 33             thread_id = coroutine.running()
 34             skynet.wait()
 35         else
 36             local msg = table.remove(message_queue,1)  //依次处理消息
 37             local session = msg.session
 38             if session == 0 then  //不需要响应
 39                 local ok, msg = do_func(f, msg)
 40                 if ok then
 41                     if msg then
 42                         skynet.fork(message_dispatch,f)
 44                     end
 45                 else
 46                     skynet.fork(message_dispatch,f)
 48                 end
 49             else
 50                 local data, size = skynet.pack(do_func(f,msg))
 51                 -- 1 means response
 52                 c.send(msg.addr, 1, session, data, size) //需要响应
 53             end
 54         end
 55     end
 56 end

上面代码实现细节不作过多分析，简单注释了下，大致就是从table数组中remove前面的消息并处理之，如果会挂起则等响应结果或超时，再处理下一条。

如上面的实现，新消息来了fork一个协程处理：

533 function skynet.fork(func,...)
534     local args = table.pack(...)  //打包参数
535     local co = co_create(function()
536         func(table.unpack(args,1,args.n)) //设置协程执行函数和参数
537     end)
538     table.insert(fork_queue, co) //回收协程资源
539     return co
540 end

104 local function co_create(f)
105     local co = table.remove(coroutine_pool)
106     if co == nil then
107         co = coroutine.create(function(...)
108             f(...)
109             while true do
110                 local session = session_coroutine_id[co]
111                 if session and session ~= 0 then
112                     local source = debug.getinfo(f,"S")
                        //log error
117                 end
118                 f = nil
119                 coroutine_pool[#coroutine_pool+1] = co
120                 f = coroutine_yield "EXIT"
121                 f(coroutine_yield())
122             end
123         end)
124     else
125         coroutine_resume(co, f)
126     end
127     return co
128 end

上面co_create就从协程池中取一个协程对象处理消息，如果没有协程对象则创建。你一定会好奇执行完后，返回结果在哪？

对于lua的协程api，当create协程时它的状态还没开始，处于挂起suspended状态，然后resume后会处理running状态，执行完后为dead状态，引用下面的：
a)coroutine.create(arg)：根据一个函数创建一个协同程序，参数为一个函数；
b)coroutine.resume(co)：使协同从挂起变为运行（1）激活coroutine，也就是让协程函数开始运行;（2）唤醒yield，使挂起的协同接着上次的地方继续运行。该函数可以传入参数；
c)coroutine.yield()：使正在运行的协同挂起，可以传入参数；

而真正强大之处在于当第二次resume时，resume和yield相关交换数据，具体怎么交互的建议看下lua协程基础。

在skynet中进行了对lua原始协程api进行封装并管理，下面说明第二个小点，当然会把第一小点也部分说明下，毕竟是个整体，从创建到处理到回收，以及中间的注意点。通过几个常用的接口来说明这套工作流程。

以下实现是wakeup相关：

493 function skynet.wakeup(token)
494     if sleep_session[token] then
495         table.insert(wakeup_queue, token) //在下一次suspend时被处理
496         return true
497     end
498 end

339 function skynet.wait(token)
340     local session = c.genid()
341     token = token or coroutine.running()
342     local ret, msg = coroutine_yield("SLEEP", session, token)//切出协程(A)
343     sleep_session[token] = nil  //协程切回来重置相关数据
344     session_id_coroutine[session] = nil
345 end

130 local function dispatch_wakeup()
131     local token = table.remove(wakeup_queue,1)
132     if token then
133         local session = sleep_session[token]
134         if session then
135             local co = session_id_coroutine[session]
136             local tag = session_coroutine_tracetag[co]
137             if tag then c.trace(tag, "resume") end
138             session_id_coroutine[session] = "BREAK"
139             return suspend(co, coroutine_resume(co, false, "BREAK"))(B)  调度被挂起的协程
140         end
141     end
142 end

157 function suspend(co, result, command, param, param2)
        //more code
183     elseif command == "SLEEP" then
184         local tag = session_coroutine_tracetag[co]
185         if tag then c.trace(tag, "sleep", co, 2) end
186         session_id_coroutine[param] = co
187         if sleep_session[param2] then
188             error(debug.traceback(co, "token duplicative"))
189         end
190         sleep_session[param2] = param
307     dispatch_wakeup()
308     dispatch_error_queue()
309 end

把要唤醒的协程通过token插入到wakeup_queue数组中（注意下，很多实现逻辑是使用table的数组部分，因为有序但带来的问题是从索引x处删除元素后，涉及到移动）

然后dispatch_wakeup会处理wakeup_queue，重点是这一句return suspend(co, coroutine_resume(co, false, "BREAK"))，这部分在后面分析。
(A)处把当前协程切出去后，那三个参数作为主协程的返回值，即coroutine_resume的返回值，再加一个本身返回的true or false，然后调用suspend，同理coroutine_resume的后两个参数作为coroutine_yield的返回值。

以上部分还是比较容易理解，这里可以结合c++协程中的实现，有专门的协程调度器，要么超时要么有数据过来（响应）进而切回相应的协程处理。

不过经历过的项目貌似没有那种加限时的请求，如果call长时间收不到响应，可能会出问题，这个需要多研究下。不过，结合skynet基础实现也好办；另外底层框架也是skynet，lua层的源码部分都有返回，不管正确还是失败都会返回，除非这条call请求消息根本没有被目标服务的消息队列收到（可能出错），或者没有被工作线程调度，再或者没有被上层服务处理；前者可能基本为零，第一种可能性不大，因为框架已经保证消息一定会被发送到消息队列中（消息队列目前是无界的），而后面两种可能确实存在，比如一个死循环或者处理耗时的功能等，这些只能靠开发人员注意及必要code review了。

311 function skynet.timeout(ti, func)
312     local session = c.intcommand("TIMEOUT",ti)
313     assert(session)
314     local co = co_create(func)
315     assert(session_id_coroutine[session] == nil)
316     session_id_coroutine[session] = co
317 end
318 
319 function skynet.sleep(ti, token)
320     local session = c.intcommand("TIMEOUT",ti)
321     assert(session)
322     token = token or coroutine.running()
323     local succ, ret = coroutine_yield("SLEEP", session, token)
324     sleep_session[token] = nil
325     if succ then
326         return
327     end
328     if ret == "BREAK" then
329         return "BREAK"
330     else
331         error(ret)
332     end
333 end

上面就是超时的实现，也即弄个协程，向skynet框架注册个定时器，当超时时，发条消息到上层，上层创建协程处理。这个跟c++协程一样，实现中不能有sleep这种调用，只能用超时，然后挂到事件列表中，超时resume协程回调，不然阻塞其他。

剩下的不过多分析，这三篇只是简单分析了个大概，还有蛮多值得学习，关键在于思考为什么要这么做，可以根据自己的经验，去尝试改进或在github上提pr，分析别人的设计，可能并不像作者一路踩坑过来，并持续重构那样，恰到好处的设计。

接下来的一篇准备研究下锁的性能，主要是对前几天学习的一个总结。

skynet 中 Lua 服务的消息处理
Lua中的协同程序 coroutine
Lua Coroutine详解
skynet 里的 coroutine
skynet coroutine 运行笔记