golang批量執(zhí)行任務(wù)的通用模板分享
需求
一個接口調(diào)用時,接收到一個列表,十個元素,需要并發(fā)執(zhí)行十個任務(wù),每個任務(wù)都要返回執(zhí)行的結(jié)果和異常,然后對返回的結(jié)果裝填到一個切片列表里,統(tǒng)一返回結(jié)果。
需要協(xié)程處理的結(jié)構(gòu)體
type Order struct {
Name string `json:"name"`
Id int `json:"id"`
}
確定通道數(shù)量
一般按入?yún)⒌男枰幚淼脑財?shù)量為準(zhǔn)
taskNum := 10
初始化通道
orderCh := make(chan Order, taskNum) //接收返回的結(jié)果 errCh := make(chan error, taskNum) //接收返回的異常
發(fā)起執(zhí)行
我們使用sync.WaitGroup來監(jiān)聽執(zhí)行情況
wg := sync.WaitGroup{}
for i:=0; i < taskNum; i++ {
wg.Add(1)
go func() {
defer wg.Done()
if i == 3 {//模擬當(dāng)i=3的時候,返回一個異常
err := errors.New("there is an error")
errCh <- err
return
}
//組裝返回結(jié)果
res := Order{
Name: "num: " + strconv.Itoa(i),
Id: i,
}
orderCh <- res
}()
}
wg.Wait() //等待所有任務(wù)執(zhí)行完畢
使用for-select接收執(zhí)行結(jié)果
orderList := make([]Order, taskNum)
for i:=0; i<taskNum; i++ {
select {
case order, ok := <-orderCh: //接收orderCh
if ok {
orderList = append(orderList, order)
}
case err := <-errCh: //接收errCh
if err != nil {
return err //看需求,這里設(shè)計發(fā)現(xiàn)一個錯誤就直接停止執(zhí)行,返回錯誤
}
default:
fmt.Println("done")
}
}
//處理完數(shù)據(jù),關(guān)閉通道
close(orderCh)
close(errCh)
超時問題
任務(wù)執(zhí)行過程中,需要控制每個任務(wù)的執(zhí)行時間,不能超過一定范圍,我們用定時器來解決這個問題
timeoutTime := time.Second * 3 //超時時間
taskTimer := time.NewTimer(timeoutTime) //初始化定時器
orderList := make([]Order, taskNum)
for i:=0; i<taskNum; i++ {
select {
....
case <-taskTimeout.C: //處理超時
err := errors.New("task timeout") //此處我們認(rèn)為超時是錯誤的一種,賦值給了err
return
...
}
//每次執(zhí)行都需要重置定時器
taskTimer.Reset(timeoutTime)
}
協(xié)程panic問題
主程序是無法捕捉協(xié)程內(nèi)的panic,因此如果不手動處理,就會發(fā)生協(xié)程內(nèi)panic導(dǎo)致整個程序中止的情況,我們在defer里處理
for i:=0; i < taskNum; i++ {
wg.Add(1)
go func() {
defer func () {
wg.Done()
//協(xié)程內(nèi)單獨捕捉異常
if r := recover(); r != nil {
err := errors.New(fmt.Sprintf("System panic:%v", r))
errCh <- err //此處將panic信息轉(zhuǎn)為err返回,也可以按需求和異常等級進行處理
return
}
}()
........
}()
}
順序問題
返回的列表元素的順序,需要跟傳參的列表順序保持一致,這時我們需要定義個帶序號的結(jié)構(gòu)體
// 需要記錄原始順序的時候,定義個帶編號的結(jié)構(gòu)體
type OrderWithSeq struct {
Seq int
OrderItem Order
}
//重寫相關(guān)排序類型
type BySeq []OrderWithSeq
func (a BySeq) Len() int {
return len(a)
}
func (a BySeq) Swap(i, j int) {
a[i], a[j] = a[j], a[i]
}
func (a BySeq) Less(i, j int) bool {
return a[i].Seq < a[j].Seq
}
// 調(diào)整返回結(jié)果
orderCh := make(chan OrderWithSeq, taskNum) //接收帶序號的結(jié)構(gòu)體
//在執(zhí)行任務(wù)時,加入序號
for i:=0; i < taskNum; i++ {
i:= i
wg.Add(1)
go func() {
····
//組裝返回結(jié)果
res := Order{
Name: "num: " + strconv.Itoa(i),
Id: i,
}
orderCh <-OrderWithSeq {
Seq: i, //帶上i這個序號
OrderItem: res,
}
}()
//接收信息,也按帶序號的結(jié)構(gòu)體進行組裝
orderSeqList := make([]OrderWithSeq, taskNum)
for i:=0; i<taskNum; i++ {
select {
case order, ok := <-orderCh: //接收orderCh
if ok {
orderList = append(orderSeqList, order)
}
.....
}
}
//按原始順序進行排序
sort.Sort(BySeq(orderSeqList))
....重新組裝數(shù)據(jù)返回
總結(jié)
標(biāo)準(zhǔn)模板如下:
type Order struct {
Name string `json:"name"`
Id int `json:"id"`
}
// 需要記錄原始順序的時候,定義個帶編號的結(jié)構(gòu)體
type OrderWithSeq struct {
Seq int
OrderItem Order
}
//重寫相關(guān)排序類型
type BySeq []OrderWithSeq
func (a BySeq) Len() int {
return len(a)
}
func (a BySeq) Swap(i, j int) {
a[i], a[j] = a[j], a[i]
}
func (a BySeq) Less(i, j int) bool {
return a[i].Seq < a[j].Seq
}
taskNum := 10
orderCh := make(chan OrderWithSeq, taskNum) //接收帶序號的結(jié)構(gòu)體
errCh := make(chan error, taskNum) //接收返回的異常
wg := sync.WaitGroup{}
//在執(zhí)行任務(wù)時,加入序號
for i:=0; i < taskNum; i++ {
i:= i
wg.Add(1)
go func() {
defer func () {
wg.Done()
//協(xié)程內(nèi)單獨捕捉異常
if r := recover(); r != nil {
err := errors.New(fmt.Sprintf("System panic:%v", r))
errCh <- err //此處將panic信息轉(zhuǎn)為err返回,也可以按需求和異常等級進行處理
return
}
}()
//組裝返回結(jié)果
res := Order{
Name: "num: " + strconv.Itoa(i),
Id: i,
}
orderCh <-OrderWithSeq {
Seq: i, //帶上i這個序號
OrderItem: res,
}
}()
wg.Wait()
//接收信息,也按帶序號的結(jié)構(gòu)體進行組裝
orderSeqList := make([]OrderWithSeq, taskNum)
timeoutTime := time.Second * 3
taskTimer := time.NewTimer(timeoutTime)
for i:=0; i<taskNum; i++ {
select {
case order, ok := <-orderCh: //接收orderCh
if ok {
orderList = append(orderSeqList, order)
}
case err := <-errCh: //接收errCh
if err != nil {
return err
}
case <-taskTimer.C: //處理超時
err := errors.New("task timeout")
return
default:
fmt.Println("done")
}
taskTimer.Reset(timeoutTime)
}
close(orderCh)
close(errCh)
//按原始順序進行排序
sort.Sort(BySeq(orderSeqList))
到此這篇關(guān)于golang批量執(zhí)行任務(wù)的通用模板分享的文章就介紹到這了,更多相關(guān)golang批量執(zhí)行任務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Go語言并發(fā)處理效率響應(yīng)能力及在現(xiàn)代軟件開發(fā)中的重要性
這篇文章主要為大家介紹了Go語言并發(fā)處理的效率及響應(yīng)能力以及在現(xiàn)代軟件開發(fā)中的重要性實例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-12-12
Go json omitempty如何實現(xiàn)可選屬性
在Go語言中,使用`omitempty`可以幫助我們在進行JSON序列化和反序列化時,忽略結(jié)構(gòu)體中的零值或空值,本文介紹了如何通過將字段類型改為指針類型,并在結(jié)構(gòu)體的JSON標(biāo)簽中添加`omitempty`來實現(xiàn)這一功能,例如,將float32修改為*float322024-09-09

