更新UDP 通知，inc_data通知change 走配置

qyxy_change_new/task.go

@@ -31,9 +31,16 @@ func IncData() {
 	wg := &sync.WaitGroup{}
 
 	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
-		if count%2000 == 0 {
+		if count%20000 == 0 {
 			log.Println("current:", count)
 		}
+		if util.ObjToString(tmp["_operation_type"]) == "update" {
+			continue
+		}
+		//废弃
+		if util.Int64All(tmp["use_flag"]) > 8 {
+			continue
+		}
 
 		zid = util.IntAll(tmp["_id"])
 		ch <- true

qyxy_es_new/main.go

@@ -101,6 +101,14 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 			log.Info("processUdpMsg", zap.Any("Unmarshal err", err))
 		}
 		log.Info("processUdpMsg", zap.Any("mapinfo", mapInfo))
+		if mapInfo != nil {
+			//相应UDP回答
+			key := utils.ObjToString(mapInfo["key"])
+			if key == "" {
+				key = "udpok"
+			}
+			go UdpClient.WriteUdp([]byte(key), udp.OP_NOOP, ra)
+		}
 
 		if tasktype, ok := mapInfo["stype"].(string); ok {
 			switch tasktype {

qyxy_es_new/task.go

@@ -36,7 +36,7 @@ func StdAdd(q interface{}) {
 	sess := Mgo.GetMgoConn()
 	defer Mgo.DestoryMongoConn(sess)
 
-	pool := make(chan bool, 10)
+	pool := make(chan bool, 18)
 	wg := &sync.WaitGroup{}
 	//q := bson.M{"_id": "affe29f8d061f3faa4170cafba41f316"}
 	//q := bson.M{"updatetime": bson.M{"$gt": Updatetime}}

qyxy_inc_data_new/config.go

@@ -11,4 +11,5 @@ type EnvConf struct {
 	Path       string
 	Addr       string
 	Dbname     string
+	ChangeUdp  bool
 }

qyxy_inc_data_new/config.toml

@@ -6,6 +6,7 @@
     path = ""
     addr = "172.17.4.181:27001"
     dbname = "mixdata"
+    changeudp = false   ## 是否通知UDP 更新企业变更信息
 
 
 

qyxy_inc_data_new/main.go

@@ -147,10 +147,13 @@ func task(path string) {
 
 	//执行完毕，通知qyxy_change，更新企业变更信息
 	data := map[string]interface{}{
-		"start":         true,
+		"start":         GF.Env.ChangeUdp,
 		"jy_updatetime": jyUpdatetime,
 	}
-	SendUdpMsg(data, changeAddr)
+
+	if GF.Env.ChangeUdp {
+		SendUdpMsg(data, changeAddr)
+	}
 
 	log.Info("task", zap.String("执行完毕", path))
 	MongoTool.Save("save_log", map[string]interface{}{"createtime": time.Now().String(), "result": saveLog})

qyxy_listen/main.go

@@ -187,6 +187,13 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 		if err != nil {
 			fmt.Println(err)
 		}
+		if mapInfo != nil {
+			key, _ := mapInfo["key"].(string)
+			if key == "" {
+				key = "udpok"
+			}
+			go UdpClient.WriteUdp([]byte(key), udp.OP_NOOP, ra)
+		}
 
 	default:
 		fmt.Println("qyxy_listen : processUdpMsg =====")