优化协程处理方式

controller/sku_controller.go

@@ -2,15 +2,16 @@ package controller
 
 
 import (
 import (
 	"encoding/json"
 	"encoding/json"
-	"github.com/gin-gonic/gin"
-	"github.com/gogf/gf/util/gconv"
-	"github.com/syyongx/php2go"
 	"go_sku_server/pkg/common"
 	"go_sku_server/pkg/common"
 	"go_sku_server/pkg/gredis"
 	"go_sku_server/pkg/gredis"
 	"go_sku_server/pkg/logger"
 	"go_sku_server/pkg/logger"
 	"go_sku_server/service"
 	"go_sku_server/service"
 	"sync"
 	"sync"
 	"time"
 	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gogf/gf/util/gconv"
+	"github.com/syyongx/php2go"
 )
 )
 
 
 const goodsSliceCount = 10 //每多少个型号id开启一个协程
 const goodsSliceCount = 10 //每多少个型号id开启一个协程
@@ -52,10 +53,11 @@ func CommonController(ctx *gin.Context) map[string]interface{} {
 		return nil
 		return nil
 	}
 	}
 
 
+	var wg sync.WaitGroup
 	ch := make(chan sync.Map) //管道
 	ch := make(chan sync.Map) //管道
-	p := 0                    //总共协程
-	zyGoodsId := make([]string, 0)
-	lyGoodsId := make([]string, 0)
+
+	zyGoodsId := make([]string, 0, goodsSliceCount)
+	lyGoodsId := make([]string, 0, goodsSliceCount)
 	for _, goodsId := range goodsIdArr {
 	for _, goodsId := range goodsIdArr {
 		if goodsId == "" {
 		if goodsId == "" {
 			continue
 			continue
@@ -65,20 +67,30 @@ func CommonController(ctx *gin.Context) map[string]interface{} {
 			if len(zyGoodsId) >= goodsSliceCount {
 			if len(zyGoodsId) >= goodsSliceCount {
 				common.PrintDebugHtml(ctx, "zy增加协程1001：")
 				common.PrintDebugHtml(ctx, "zy增加协程1001：")
 				common.PrintDebugHtml(ctx, zyGoodsId)
 				common.PrintDebugHtml(ctx, zyGoodsId)
-				//wg.Add(1) //协程计数一
-				go zyService.ZyGoodsDetail(ctx, zyGoodsId, ch)
-				zyGoodsId = zyGoodsId[:0:0]
-				p++
+				wg.Add(1) //协程计数一
+
+				// 必须创建一个新的slice副本，否则数据会被覆盖
+				idsToProcess := make([]string, len(zyGoodsId))
+				copy(idsToProcess, zyGoodsId)
+				go func() {
+					defer wg.Done()
+					zyService.ZyGoodsDetail(ctx, idsToProcess, ch)
+				}()
+				zyGoodsId = zyGoodsId[:0]
 			}
 			}
 		} else { //联营
 		} else { //联营
 			lyGoodsId = append(lyGoodsId, goodsId)
 			lyGoodsId = append(lyGoodsId, goodsId)
 			if len(lyGoodsId) >= goodsSliceCount {
 			if len(lyGoodsId) >= goodsSliceCount {
 				common.PrintDebugHtml(ctx, "ly增加协程1002：")
 				common.PrintDebugHtml(ctx, "ly增加协程1002：")
 				common.PrintDebugHtml(ctx, lyGoodsId)
 				common.PrintDebugHtml(ctx, lyGoodsId)
-				//wg.Add(1)
-				go lyService.LyGoodsDetail(ctx, lyGoodsId, ch)
-				lyGoodsId = lyGoodsId[:0:0]
-				p++
+				wg.Add(1)
+				idsToProcess := make([]string, len(lyGoodsId))
+				copy(idsToProcess, lyGoodsId)
+				go func() {
+					defer wg.Done()
+					lyService.LyGoodsDetail(ctx, idsToProcess, ch)
+				}()
+				lyGoodsId = lyGoodsId[:0]
 			}
 			}
 		}
 		}
 	}
 	}
@@ -86,35 +98,53 @@ func CommonController(ctx *gin.Context) map[string]interface{} {
 	if len(zyGoodsId) > 0 {
 	if len(zyGoodsId) > 0 {
 		common.PrintDebugHtml(ctx, "zy增加协程1003：")
 		common.PrintDebugHtml(ctx, "zy增加协程1003：")
 		common.PrintDebugHtml(ctx, zyGoodsId)
 		common.PrintDebugHtml(ctx, zyGoodsId)
-		//wg.Add(1) //协程计数一
-		go zyService.ZyGoodsDetail(ctx, zyGoodsId, ch)
-		p++
+		wg.Add(1) //协程计数一
+		idsToProcess := make([]string, len(zyGoodsId))
+		copy(idsToProcess, zyGoodsId)
+		go func() {
+			defer wg.Done()
+			zyService.ZyGoodsDetail(ctx, idsToProcess, ch)
+		}()
 	}
 	}
 
 
 	if len(lyGoodsId) > 0 {
 	if len(lyGoodsId) > 0 {
 		common.PrintDebugHtml(ctx, "ly增加协程1004：")
 		common.PrintDebugHtml(ctx, "ly增加协程1004：")
-		common.PrintDebugHtml(ctx, zyGoodsId)
-		go lyService.LyGoodsDetail(ctx, lyGoodsId, ch)
-		p++
+		common.PrintDebugHtml(ctx, lyGoodsId)
+		wg.Add(1)
+		idsToProcess := make([]string, len(lyGoodsId))
+		copy(idsToProcess, lyGoodsId)
+		go func() {
+			defer wg.Done()
+			lyService.LyGoodsDetail(ctx, idsToProcess, ch)
+		}()
 	}
 	}
 
 
+	// 开启一个协程，等待所有任务完成，然后关闭channel
+	go func() {
+		wg.Wait()
+		close(ch)
+	}()
+
 	//异步map最后转成map
 	//异步map最后转成map
 	temp := make(map[string]interface{})
 	temp := make(map[string]interface{})
-	for i := 0; i < p; i++ {
+	timeout := time.After(time.Second * 20)
+	for {
 		select {
 		select {
-		case GoodsRes := <-ch:
+		case GoodsRes, ok := <-ch:
+			if !ok { // channel被关闭，说明所有协程都已执行完毕
+				return temp
+			}
 			GoodsRes.Range(func(k, v interface{}) bool {
 			GoodsRes.Range(func(k, v interface{}) bool {
 				s, _ := k.(string)
 				s, _ := k.(string)
 				temp[s] = v
 				temp[s] = v
 				return true
 				return true
 			})
 			})
 
 
-		case <-time.After(time.Second * 20):
-			logger.Log("协程超时", "sku", 1)
+		case <-timeout:
+			logger.Log("协程整体处理超时", "sku", 1)
+			return temp // 超时，返回已经收到的部分数据
 		}
 		}
 	}
 	}
-	return temp
-
 }
 }
 func Synchronization(ctx *gin.Context) {
 func Synchronization(ctx *gin.Context) {
 
 
@@ -130,7 +160,7 @@ func Hbsdata(ctx *gin.Context) {
 }
 }
 
 
 /*
 /*
- 测试redis
+测试redis
 */
 */
 func Testr(ctx *gin.Context) {
 func Testr(ctx *gin.Context) {