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}
}//获取执行完的error
func (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内存储了三分子相同的数据，而切时最后那次循环的数据。
这个问题今天目前还未解决。对于新手真的很无奈。后续还会跟进，也求助各位大佬帮忙。