jy-data-extract/fullproject/src_dev4/main.go

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)
		}
	}
}