在线时间:8:00-16:00
迪恩网络APP
随时随地掌握行业动态
扫描二维码
关注迪恩网络微信公众号
本文微信公众号文章:https://mp.weixin.qq.com/s/aHJScjcrrJyYlIDcA00P0g 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日
如果你觉得本文对你有帮助,可以转到你的朋友圈,让更多人一起学习。 第一时间获取文章,可以关注本人公众号:月牙寂道长,也可以扫码关注
|
请发表评论