爱站程序员基地go异步处理大量数据，并发-Runner，封装多个任务
task/runner_async.go
package taskimport (\"os\"\"os/signal\"\"sync\"\"time\")//异步执行任务type Runner struct {//操作系统的信号检测interrupt chan os.Signal//记录执行完成的状态complete chan error//超时检测timeout <-chan time.Time//保存所有要执行的任务,顺序执行tasks []func(id int) errorwaitGroup sync.WaitGrouplock sync.Mutexerrs []error}//new一个Runner对象func NewRunner(d time.Duration) *Runner {return &Runner{interrupt: make(chan os.Signal, 1),complete: make(chan error),timeout: time.After(d),waitGroup: sync.WaitGroup{},lock: sync.Mutex{},}}//添加一个任务func (this *Runner) Add(tasks ...func(id int) error) {this.tasks = append(this.tasks, tasks...)}//func (this *Runner) Add(tasks []func(id int) error) {// this.tasks = tasks//}//启动Runner，监听错误信息func (this *Runner) Start() error {//接收操作系统信号signal.Notify(this.interrupt, os.Interrupt)//并发执行任务go func() {this.complete <- this.Run()}()select {//返回执行结果case err := <-this.complete:return err//超时返回case <-this.timeout:return ErrTimeout}}//异步执行所有的任务func (this *Runner) Run() error {for id, task := range this.tasks {if this.gotInterrupt() {return ErrInterrupt}this.waitGroup.Add(1)go func(id int) {this.lock.Lock()//执行任务err := task(id)//加锁保存到结果集中this.errs = append(this.errs, err)this.lock.Unlock()this.waitGroup.Done()}(id)}this.waitGroup.Wait()return nil}//判断是否接收到操作系统中断信号func (this *Runner) gotInterrupt() bool {select {case <-this.interrupt://停止接收别的信号signal.Stop(this.interrupt)return true//正常执行default:return false}}//获取执行完的errorfunc (this *Runner) GetErrs() []error {return this.errs}
task/err.go
package taskimport \"errors\"//超时错误var ErrTimeout = errors.New(\"received timeout\")//操作系统系统中断错误var ErrInterrupt = errors.New(\"received interrupt\")
####### 测试示例 task/runner_async_test.go
package taskimport (\"gm_server/db\"\"gm_server/models\"\"time\"\"fmt\"\"os\"\"runtime\")func RestoreRunnerStart(cardId int, data []models.TblCard) {//开启多核心runtime.GOMAXPROCS(runtime.NumCPU())//创建runner对象，设置超时时间runner := NewRunner(18 * time.Second)//total := len(data)/3 + 1//start := 0//end := 3//var tasks []func(id int) error//for i := 0; i < total; i++ {// if end > len(data) {// end = len(data)// }// tasks = append(tasks,createRestoreTask(cardId, data[start:end]))// fmt.Println(tasks)// //添加运行的任务// start += 3// end += 3//}////runner.Add(// tasks,//)runner.Add(createRestoreTask(cardId, data),//createTask(),//createTask(),//createTask(),)fmt.Println(\"异步执行任务\")//开始执行任务if err := runner.Start(); err != nil {switch err {case ErrTimeout:fmt.Println(\"执行超时\")os.Exit(1)case ErrInterrupt:fmt.Println(\"任务被中断\")os.Exit(2)}}fmt.Println(\"执行结束\")}//创建要执行的任务func createRestoreTask(cardId int, data []models.TblCard) func(id int) error {return func(id int) error {fmt.Printf(\"正在执行%v个任务\\n\", id+1)fmt.Printf(\"一共%v条数据\\n\", len(data))fmt.Println(data)//模拟任务执行,sleep//time.Sleep(1 * time.Second)turn nil}}
Add添加一个任务，任务为接收int类型的一个闭包
createRestoreTask为创建的任务，在runner.add中可重复添加，为同时执行多个任务
在这里我传参过来的是data，为数据库查询出来的数据，目的想拆分为多个任务进行处理，所以使用了切片分割数据，append进tasks内，行成一个整体的参数在run.add中使用，
也就是代码中我将数据分为三份，循环了三次，这三分数据依次调用createRestoreTask方法并存储在tasks中。
但是这里遇到了一个问题，切片是引用传值，当使用append时没有改变内存地址，导致我循环了三次后，每个元素内存贮的都是最后一次循环的结果，覆盖了之前循环的数据。也就是tasks内存储了三分子相同的数据，而切时最后那次循环的数据。
这个问题今天目前还未解决。对于新手真的很无奈。后续还会跟进，也求助各位大佬帮忙。
