并发问题

controller/sku_controller.go

@@ -62,6 +62,10 @@ func CommonController(ctx *gin.Context) map[string]interface{} {
 		common.Output(ctx, 1001, "查询型号ID不得为空", "")
 		return nil
 	}
+	if len(goodsIdArr) > 500 {
+		common.Output(ctx, 1001, "查询型号ID不得超过500个", "")
+		return nil
+	}
 
 	//////////初始化公共变量 以及 批量查询///////////
 	zyGoodsId := make([]string, 0, goodsSliceCount)
@@ -116,18 +120,13 @@ func CommonController(ctx *gin.Context) map[string]interface{} {
 	/////////////////多线程///////////////////////////
 
 	var wg sync.WaitGroup
-	ch := make(chan sync.Map, 50) //管道
-	//wgMax := 3                    //线程数量最多多少
-	//wgcount := 0 //当前线程数
+	ch := make(chan sync.Map, 500)       //管道
+	semaphore := make(chan struct{}, 10) // 限制最大并发数为10
 
 	temp := make(map[string]interface{}) //最后输出计算结果
 	sd := 0
 	for _, goodsId := range goodsIdArr {
 		sd += 1
-		//if len(goodsIdArr) > 100 {
-		//	common.Output(ctx, 1001, "查询型号ID不得超过100个", "")
-		//	return nil
-		//}
 		if goodsId == "" {
 			continue
 		}
@@ -136,7 +135,8 @@ func CommonController(ctx *gin.Context) map[string]interface{} {
 			if len(zyGoodsId) >= goodsSliceCount {
 				common.PrintDebugHtml(ctx, "zy增加协程1001：")
 				common.PrintDebugHtml(ctx, zyGoodsId)
-				wg.Add(1) //协程计数一
+				wg.Add(1)               //协程计数一
+				semaphore <- struct{}{} // 获取信号量
 
 				idsToProcess := make([]string, len(zyGoodsId))
 				copy(idsToProcess, zyGoodsId)
@@ -144,6 +144,8 @@ func CommonController(ctx *gin.Context) map[string]interface{} {
 				// 启动协程，传递独立的 context 和参数，而不是 gin.Context
 				go func(ctx context.Context, params service.RequestParams, goodsIds []string, ch chan sync.Map) {
 					defer wg.Done()
+					defer func() { <-semaphore }() // 释放信号量
+
 					zyService.ZyGoodsDetail(ctx, params, goodsIds, ch)
 				}(ctx, requestParams, idsToProcess, ch)
 
@@ -163,37 +165,30 @@ func CommonController(ctx *gin.Context) map[string]interface{} {
 				temp[goodsId] = false
 				continue
 			}
-
+			wg.Add(1)
+			semaphore <- struct{}{} // 获取信号量
 			// 启动协程，传递独立的 context 和参数，而不是 gin.Context
 			go func(ctx context.Context, params service.RequestParams, skuStr, goodsId, spuStr string, ch chan sync.Map, tag int) {
-				//defer wg.Done()
-				defer func() { <-ch }()
+				defer wg.Done()
+				defer func() { <-semaphore }() // 释放信号量
+
 				lyService.LyGoodsDetail(ctx, params, skuStr, goodsId, spuStr, ch, tag)
 			}(ctx, requestParams, skuStr, goodsId, redisLySpuArr[spuId], ch, sd)
 
-			//wg.Add(1)
-			//	wgcount += 1
-
-			//if wgcount >= wgMax {
-			//	println("线程数量:" + gconv.String(wgcount))
-			//	wg.Wait() // 等待所有登记的goroutine都结束
-			//	wgcount = 0
-			//}
 		}
 	}
 
-	//if wgcount > 0 {
-	//	//println("等待结束,现有运行线程数：" + gconv.String(wgcount))
-	//	//wg.Wait()
-	//}
-	//wg.Wait()
+	// 启动通道关闭器
+	go func() {
+		wg.Wait()
+		close(ch)
+	}()
 
 	//异步map最后转成map
 	for {
 		select {
 		case GoodsRes, ok := <-ch:
 			if !ok { // channel被关闭，说明所有协程都已执行完毕或超时
-				close(ch)
 				return temp
 			}
 			GoodsRes.Range(func(k, v interface{}) bool {
@@ -209,8 +204,8 @@ func CommonController(ctx *gin.Context) map[string]interface{} {
 	}
 }
 func Synchronization(ctx *gin.Context) {
-
 	res := CommonController(ctx)
+	println(len(res))
 	common.Output(ctx, 0, "success", res)
 }
 

service/service_ly.go

@@ -8,12 +8,10 @@ import (
 	"go_sku_server/pkg/logger"
 	"go_sku_server/pkg/mongo"
 	"go_sku_server/service/sorter"
-	"sort"
-	"sync"
-	"time"
-
 	"gopkg.in/mgo.v2"
 	"gopkg.in/mgo.v2/bson"
+	"sort"
+	"sync"
 
 	"github.com/gogf/gf/util/gconv"
 	"github.com/gomodule/redigo/redis"
@@ -39,7 +37,7 @@ type Power struct {
 */
 func (ls *LyService) LyGoodsDetail(ctx context.Context, params RequestParams, skuStr, goodsId, spuStr string, ch chan sync.Map, tag int) {
 
-	start_time := gconv.String(time.Now().UnixMilli())
+	//start_time := gconv.String(time.Now().UnixMilli())
 	redisConn := gredis.Conn("search_r")
 	redisConnSpu := gredis.Conn("spu")
 	// 连接prev_sku MongoDB
@@ -275,7 +273,7 @@ func (ls *LyService) LyGoodsDetail(ctx context.Context, params RequestParams, sk
 
 	//退出通道
 	ch <- GoodsRes
-	println("---tag:" + gconv.String(tag) + "-----goodsId:" + goodsId + "--start_time:" + start_time + " ---end_time:" + gconv.String(time.Now().UnixMilli()))
+	//println("---tag:" + gconv.String(tag) + "-----goodsId:" + goodsId + "--start_time:" + start_time + " ---end_time:" + gconv.String(time.Now().UnixMilli()))
 }
 
 // 获取活动