udp验证

udp_winner/main.go

@@ -112,21 +112,42 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 			}
 			//data_info:save//存量   data_info:add //增量
 			//阻塞
-			CPoolWinner <- true
-			go func(mapinfo *map[string]interface{}) {
-				defer func() { <-CPoolWinner }()
-				TaskWinner(mapinfo)
-			}(tmp)
-			CPoolBuery <- true
-			go func(mapinfo *map[string]interface{}) {
-				defer func() { <-CPoolBuery }()
-				TaskBuyer(mapinfo)
-			}(tmp)
-			CPoolAgency <- true
-			go func(mapinfo *map[string]interface{}) {
-				defer func() { <-CPoolAgency }()
-				TaskAgency(mapinfo)
-			}(tmp)
+			tmpstype,ok := (*tmp)["stype"].(string)
+			if ok&& tmpstype ==""{
+				CPoolWinner <- true
+				go func(mapinfo *map[string]interface{}) {
+					defer func() { <-CPoolWinner }()
+					TaskWinner(mapinfo)
+				}(tmp)
+				CPoolBuery <- true
+				go func(mapinfo *map[string]interface{}) {
+					defer func() { <-CPoolBuery }()
+					TaskBuyer(mapinfo)
+				}(tmp)
+				CPoolAgency <- true
+				go func(mapinfo *map[string]interface{}) {
+					defer func() { <-CPoolAgency }()
+					TaskAgency(mapinfo)
+				}(tmp)
+			}else if tmpstype =="winner" {
+				CPoolWinner <- true
+				go func(mapinfo *map[string]interface{}) {
+					defer func() { <-CPoolWinner }()
+					TaskWinner(mapinfo)
+				}(tmp)
+			}else if tmpstype=="buyer"{
+				CPoolBuery <- true
+				go func(mapinfo *map[string]interface{}) {
+					defer func() { <-CPoolBuery }()
+					TaskBuyer(mapinfo)
+				}(tmp)
+			}else if tmpstype=="agency"{
+				CPoolAgency <- true
+				go func(mapinfo *map[string]interface{}) {
+					defer func() { <-CPoolAgency }()
+					TaskAgency(mapinfo)
+				}(tmp)
+			}
 
 		}
 	case mu.OP_NOOP: //下个节点回应

udp_winner/timedTaskWinner.go

@@ -141,7 +141,7 @@ func TaskWinner(mapinfo *map[string]interface{}) {
 				"gtid":  tmpRangeId,
 				"lteid": lteid,
 				"data_info":"save",
-				"stype": "",
+				"stype": "winner",
 			})
 			if e := udpclient.WriteUdp(by, util2.OP_TYPE_DATA, &net.UDPAddr{
 				IP:   net.ParseIP("127.0.0.1"),