Golang 异步任务执行器——Gochan
敬维 · · 6284 次点击 · · 开始浏览写在前面
在最近编码过程中,大量使用了异步任务。在自己需求的基础上抽象出一个异步任务执行器,应该有挺多类似的需求,于是开源出来。项目地址为《GitHub - chalvern/gochan》,还希望大家能够不吝 star ✨。
项目背景
一般情况下,我们可以通过定义一个带缓冲的 channel 变量接收某种事件,然后通过一个专用的 goroutine 消费执行这个 channel 中的事件。
但是如果相关事件很多的时候,一个 goroutine 不够用了怎么办呢?或许我们会想到多创建几个专用的 goroutine 来并发地消费执行这个 channel 中的事件;如果 channel 中各个事件之间是独立的,是可行的,但是如果某些事件之间具有某种顺序上的约束,那么就需要对事件进行特定的分类。
比如,一个订单的支付与货物发货,两个事件是需要保序的;但是不同的订单之间又是可以并发执行;其实就是实现一个微型的按特定主题分类的 pub-sub(发布-订阅)系统。以订单为例,可以根据订单单号,把相同单号的事件推送到同一个队列(channel),一个特定的执行器(goroutine)来消费执行这个队列中的事件,如此平行扩展多个类似的组合,实现并发。
设计的思路
平常的设计
通过定义一个带缓冲的 channel 变量接收某种事件,然后通过一个专用的 goroutine 消费执行这个 channel 中的事件。
event ->
|
event -> buffer-channel -> goroutine
|
event ->
状态无依赖的并发设计
事件之间完全没有状态依赖,因此可以简单扩展 goroutine 进行加快事件执行速度。
event -> ->goroutine
| |
event -> buffer-channel -> goroutine
| |
event -> ->goroutine
状态存在依赖的并发设计 (gochan)
引入一层分发器(dispatcher),根据某个特性(比如 uuid)把事件分发到相应的队列(buffer-channel)中。
event -> -> buffer-channel -> goroutine
| |
event --> dispatcher -> buffer-channel -> goroutine
| |
event -> -> buffer-channel -> goroutine
使用示例
packagemainimport("errors""math/rand""time""github.com/chalvern/gochan")typeManagerstruct{gochanNumintbufferNumintdispatcher*gochan.Dispatcher}func(m*Manager)Dispatch(objIDint,taskgochan.TaskFunc)error{ifobjID<0{objID=rand.Intn(m.gochanNum)}returnm.dispatcher.Dispatch(objID,task)}func(m*Manager)Close(){m.dispatcher.Close()}funcmain(){gochanNum:=3bufferNum:=10manager:=Manager{gochanNum:gochanNum,bufferNum:bufferNum,dispatcher:gochan.NewDispatcher(gochanNum,bufferNum),}objID:=1task1:=func()error{returnerrors.New("task 1")}manager.Dispatch(objID,task1)time.Sleep(time.Second)}参考
有疑问加站长微信联系(非本文作者)
入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889
关注微信- 请尽量让自己的回复能够对别人有帮助
- 支持 Markdown 格式, **粗体**、~~删除线~~、
`单行代码` - 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
- 图片支持拖拽、截图粘贴等方式上传
收入到我管理的专栏 新建专栏
写在前面
在最近编码过程中,大量使用了异步任务。在自己需求的基础上抽象出一个异步任务执行器,应该有挺多类似的需求,于是开源出来。项目地址为《GitHub - chalvern/gochan》,还希望大家能够不吝 star ✨。
项目背景
一般情况下,我们可以通过定义一个带缓冲的 channel 变量接收某种事件,然后通过一个专用的 goroutine 消费执行这个 channel 中的事件。
但是如果相关事件很多的时候,一个 goroutine 不够用了怎么办呢?或许我们会想到多创建几个专用的 goroutine 来并发地消费执行这个 channel 中的事件;如果 channel 中各个事件之间是独立的,是可行的,但是如果某些事件之间具有某种顺序上的约束,那么就需要对事件进行特定的分类。
比如,一个订单的支付与货物发货,两个事件是需要保序的;但是不同的订单之间又是可以并发执行;其实就是实现一个微型的按特定主题分类的 pub-sub(发布-订阅)系统。以订单为例,可以根据订单单号,把相同单号的事件推送到同一个队列(channel),一个特定的执行器(goroutine)来消费执行这个队列中的事件,如此平行扩展多个类似的组合,实现并发。
设计的思路
平常的设计
通过定义一个带缓冲的 channel 变量接收某种事件,然后通过一个专用的 goroutine 消费执行这个 channel 中的事件。
event ->
|
event -> buffer-channel -> goroutine
|
event ->
状态无依赖的并发设计
事件之间完全没有状态依赖,因此可以简单扩展 goroutine 进行加快事件执行速度。
event -> ->goroutine
| |
event -> buffer-channel -> goroutine
| |
event -> ->goroutine
状态存在依赖的并发设计 (gochan)
引入一层分发器(dispatcher),根据某个特性(比如 uuid)把事件分发到相应的队列(buffer-channel)中。
event -> -> buffer-channel -> goroutine
| |
event --> dispatcher -> buffer-channel -> goroutine
| |
event -> -> buffer-channel -> goroutine
使用示例
packagemainimport("errors""math/rand""time""github.com/chalvern/gochan")typeManagerstruct{gochanNumintbufferNumintdispatcher*gochan.Dispatcher}func(m*Manager)Dispatch(objIDint,taskgochan.TaskFunc)error{ifobjID<0{objID=rand.Intn(m.gochanNum)}returnm.dispatcher.Dispatch(objID,task)}func(m*Manager)Close(){m.dispatcher.Close()}funcmain(){gochanNum:=3bufferNum:=10manager:=Manager{gochanNum:gochanNum,bufferNum:bufferNum,dispatcher:gochan.NewDispatcher(gochanNum,bufferNum),}objID:=1task1:=func()error{returnerrors.New("task 1")}manager.Dispatch(objID,task1)time.Sleep(time.Second)}