protoactor-go 源码分析：async schedule

actor是一种异步并发处理模型。最具代表性的是erlang语言。

在golang中，最具代表性的并发模式为csp，多协程并发。

这两者的区别更多类似于网络并发。 网络程序设计中的并发复杂性 这篇文章中有介绍，关于事件驱动和多线程并发。其两者都有着各自的优缺点。

protoactor-go的源码地址为：

https://github.com/AsynkronIT/protoactor-go

对于一个大型的项目来讲，我个人的学习习惯于从最小版本开始学起。这是因为，在一个项目最初的时候，大体功能和架构都已经成形，最初的版本，一般来说，代码量都较少，功能集最小。学习曲线低，并且又最初版本，慢慢往高版本过渡，也能更了解项目进化的过程，也是一个学习的过程。

并且在实际使用过程中，大多数情况下，我们可能不需要那么多的功能集，并且需要根据实际情况做一些二次开发，此时的话，也许低版本的会更贴近实际使用场景和二次开发场景。

基本的模块划分，有最初版本的源码分析：

开源代码protoactor-go[e866f39]源码分析

下面分析2661d6a版本

此版本相对于上一篇文章 开源代码protoactor-go[e866f39]源码分析 来说，增加了一个异步调度。其余地方都是一样的。

/github.com/AsynkronIT/protoactor-go/main.go

流程图中发生变化的地方：

在ChannelActorRef的Tell，SendSystemMessage，中增加了调用schedule

ActorzOf，现在只承担初始化操作，并没有调度的功能。

上面是两对变量，用于原子变量操作。

64：这里的调度，会先通过原子操作CompareAndSwapInt32，来做同步的判断，判断schedulerStatus是否为Idle，是则swap成Busy，否则不swap。这个是常用的原子同步操作

65：因为有人调用schedule，说明是有msg需要传递。这里将hasMoreMessages设置成MailboxHasMoreMessages

66：判断是否有swap，防止重入

67：开启processMessages协程

处理msg协程

72：防止重入，将hasMoreMessages设置空（MailBoxHasNoMessages）

73-86：一个重复30次的for循环，尝试30次的操作，进行msg的chan监听，并调用相应的处理接口。

87：操作完，将schedulerStatus设置为空（MailboxIdle ）

88-95：属于double check的作用，防止在86-87行代码之间有人调用schedule而被忽略掉。这个是关键点。

88：读取hasMoreMessages信息，看看是否有没有处理的信息

89：读取scheduleStatus信息

90：判断是否有未读取信息，并判断status是否为空（MailboxIdle）

91：尝试将schedulerStatus状态进行CompareAndSwap操作

93：重启一个新的processMessages协程。

这里的异步调度操作，利用了原子操作来实现同步，并采用double check方式来防止信息忽略。这个调度处理是很值得学习的。可以用到很多地方。

那么我们再看看最新的版本里调度是如何的？

代码最新更新是2019年4月29日 GMT+8 下午11:29:31

代码目录：

从这里也可以看到随着版本的升级，相对比最初版本，代码复杂度越来越高，模块抽象越来越多，当然功能也越来越多。

但原理其实没差。

我们看看最新版本的调度

github.com/AsynkronIT/protoactor-go/mailbox/mailbox.go

模块做了抽象，上面的

MessagesInvoker：消息真正处理的地方

Mailbox：消息调度的地方

有一个default的Mailbox，暂时还没有看到其他的Mailbox。那么就以这个来分析

在看之前，我们先看下dispatcher

github.com/AsynkronIT/protoactor-go/mailbox/dispatcher.go

这里有异步和同步两个dispatcher

goroutineDispatcher：异步的，在Schedule中，开启了协程

synchronizeDispatcher：同步的，在Schedule中，是同步执行

这两个type都是int，在throughput返回的其int值，这个将在调度中有作用

回到mailbox中

注册接口：消息处理接口，Dispatcher接口（异步和同步）

两个消息发送：

1、postUserMessage，发送user message

2、postSystemMessage，发送system message

从处理流程上，和最初的调度版本来看，流程是一样的。都在消息发送中，调用了schedule

这里依旧用的原子操作来进行同步，对schedulerStatus进行CompareAndSwapInt32操作。

然后就进如到Dispatcher中的Schedule，这个取决于Dispatcher初始化是异步还是同步，提供了goroutineDispatcher和synchronizeDispatcher供选择。

85：真正的处理，封装到run中了。

88：将SchedulerStatus设置为空（idle）

89：加载sysmessages数量

90：加载usermessages数量

92：判断是否有消息还未处理

94：重新check schedulerStatus，这里就是double check

96：若还有未处理的，重新进入run

整个处理流程是一模一样的，并没有发生变化

看下run中

115：设置了i为0，t为Dispatcher的值。这里是与之前调度不一样的地方

117-120：当i>t时候，调用了runtime.Gosched（）。此处是与最初版本不一样的地方。增加这个的好处是，当处理的消息数量过多时，这个for循环是不停的运行。增加了Gosched的话，是当消息处理一定数量的时候，让其停止运行，等待runtime重新调度其运行。这样可以防止cpu资源一直被消息处理占用。

125-139：systemMessage的处理地方

146-152：userMessage的处理地方

从最初版本，与最新版本的调度对比来看。其原理几乎没有发生变化。也验证了，对于开源代码的学习来说，学习原理的话，并非是最高版本是好。反而低版本对学习原理来说，学习上手更快。

龚浩华

月牙寂道长

qq：29185807

2019年05月15日

如果你觉得本文对你有帮助，可以转到你的朋友圈，让更多人一起学习。

第一时间获取文章，可以关注本人公众号：月牙寂道长，也可以扫码关注