Golang controls the number of goroutines and obtains processing results

One 、 Preface

       Recently encountered batch refresh ES The need for data , To speed up processing , That must be the first choice goroutine 了 , But as we all know ,goroutine The return value and error handling of have always been elusive ,go Get out easy , How to monitor go The result of going out is a problem .

1、goroutine Error handling of

      sync.ErrGroup stay sync.WaitGroup On the basis of function , Added error passing , And cancel the whole... In case of an unrecoverable error goroutine aggregate , Or wait timeout .

You can learn more about Baidu .errGroup

2、goroutine Processing results of

       Currently in use goroutine It is generally used that channel 、 sync.WaitGroup 、context, To realize the process control and message transmission between various processes , The first choice is channel To get the processing results ,channel Reference resources ：go adopt channel obtain goroutine Processing results of .

except channel, So can we use concurrent security sync.Map To store the results , Of all the goroutine After execution , Get the processing results in a unified way ？ The answer is yes ,sync.Map It can be perfectly realized .

Two 、 control goroutine Quantity and obtain processing results

1、 Combat code

Here are the USES goroutine Batch refresh ES data , And get the code of the processing result ：

// Define the return value to get , Number of successes , The number of failures , Failure id aggregate 
succeededNums, failedNums:= 0, 0, 
errData:=""
var syncMap sync.Map
wg := sync.WaitGroup{
    }
// control goroutine Number , At the same time, only 10 individual goroutine
chan1 := make(chan struct{
    }, 10) 
for {
    
   // Business logic 
   // ....
   //goroutine Speed up ,chan1 Write full , The block . Waiting for the previous goroutine Release to continue the cycle 
   chan1 <- struct{
    }{
    }
   wg.Add(1)
   go func(ctx context.Context, targetIndexName string, esClient *elastic.Client) {
    
      defer func() {
    
         if err1 := recover(); err1 != nil {
     // Produced panic abnormal 
            log.Errorf(ctx, "%s go panic! err:(+%v)", logPreFix, err1)
         }
         wg.Done() // Every goroutine Release after execution 
         return
      }()
      bulkRequest := esClient.Bulk()
      //ES Use bulk Method batch refresh data 
      bulkResByAssetInfo := new(EsBulkDataRes)
      bulkResByAssetInfo, err = BulkEsDataByAssetInfo(ctx, bulkRequest, targetIndexName)
      if err != nil {
    
         log.Errorf(ctx, "%s BulkEsDataByAssetInfo error (%+v) ,test:(+%v)", logPreFix, err, string_util.TransferToString(fileUnitList))
         return
      }
      // Accumulate the execution results to sync.map, Ensure concurrent security 
      tempMap := make(map[string]interface{
    })
      tempMap["successNums"] = bulkResByAssetInfo.SucceededNums
      tempMap["failedNums"] = bulkResByAssetInfo.FailedNums
      tempMap["errData"] = bulkResByAssetInfo.ErrData
      // Take the maximum of each cycle id, As syncMap Of key
      syncMap.Store(test[nums-1].ID, tempMap)
      // Release after execution channel
      <-chan1
   }(ctx, targetIndexName, esClient)
  
}
wg.Wait()
// Refresh end , Write notification , The Notice includes , Traverse sync.map, Get the return value 
syncMap.Range(func(key, value interface{
    }) bool {
    
   val := value.(map[string]interface{
    })
   succeededNums += val["successNums"].(int)
   failedNums += val["failedNums"].(int)
   errData += val["errData"].(string)
   return true
})

// Print the results 
fmt.Println(" Number of successes :",succeededNums)
fmt.Println(" The number of failures :",failedNums)
fmt.Println(" Failure id:",errData)

2、syncMap Use

（1） Write the processing result to map
（2）  write in map To sync.Map in , Be careful key Don't repeat 
（3） Use Range To cycle sync.Map, Get the results , Sum up 

3、 control goroutine The number of

This is mainly through setting channel Length to achieve .

（1） Set up channel length , The cycle starts with each generation goroutine Write once channel
（2） channel If it is full, it will block 
（3）goroutine completion of enforcement , Release channel
（4） for Continue writing in the loop channel, Ensure simultaneous execution goroutine Only 10 individual 

3、 ... and 、sync.Map The shortcomings of

1、 Need to be right value Do assertion processing , This is interface{} The nature of

2、 Everybody knows sync.Map Suitable for reading more and writing less , Bloggers here are running scripts , So use sync.Map It doesn't hurt , If you want to pursue performance , You can look at it currentMap The implementation of the , adopt hash Points barrels , Reduce the granularity of locks to improve performance .
current-Map

end