package main import ( "encoding/json" "log" mu "mfw/util" "net" "time" ) var ( udpclient mu.UdpClient //udp对象 SingleThread = make(chan bool, 1) ) func main() { //udp跑增量 id段 zl //udp跑全量 ql //udp跑历史数据 信息id1,id2/或id段 ls //udp强制合并 信息id1,id2,id3 [项目id] 不存在时新建 qzhb //udp强制拆分 项目id,信息id1,id2 qzcf //udp重新合并 信息id1,id2,id3 cxhb //loadData("project_0809", 1, true) time.Sleep(99999 * time.Hour) } func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) { switch act { case mu.OP_TYPE_DATA: //上个节点的数据 var mapInfo map[string]interface{} err := json.Unmarshal(data, &mapInfo) log.Println("err:", err, "mapInfo:", mapInfo) if err != nil { udpclient.WriteUdp([]byte("err:"+err.Error()), mu.OP_NOOP, ra) } else if mapInfo != nil { key, _ := mapInfo["key"].(string) if key == "" { key = "udpok" } go udpclient.WriteUdp([]byte(key), mu.OP_NOOP, ra) SingleThread <- true tasktype, _ := mapInfo["stype"].(string) log.Println("tasktype:", tasktype) switch tasktype { case "ql": go func() { defer func() { <-SingleThread }() taskQl(mapInfo) }() case "ids": go func() { defer func() { <-SingleThread }() Ids(mapInfo) }() } } case mu.OP_NOOP: //下个节点回应 ok := string(data) if ok != "" { log.Println("ok:", ok) } } }