全量融合-指定有效字段-重构

udpfusion/src/config.json

@@ -3,7 +3,7 @@
   "mongodb": {
     "addrName": "192.168.3.207:27092",
     "dbName": "zhengkun",
-    "collName": "all_01_02_fusiontest",
+    "collName": "fusion_test",
     "pool": 20,
     "mgo_pool": 10,
     "site": {
@@ -12,7 +12,7 @@
     }
   },
   "es": {
-    "addr": "http://127.0.0.1:12003",
+    "addr": "http://192.168.3.11:9800",
     "size":20,
     "es_pool": 10,
     "index": "zktest",
@@ -29,48 +29,37 @@
   "nextNode": [
 
   ],
-  "notFusionKey": {
-    "title":0,
-    "detail":0,
-    "href":0,
-    "contenthtml": 0,
-    "summary": 0,
-    "publishtime": 0,
-    "spidercode": 0,
-    "site": 0,
-    "channel": 0,
-    "comeintime": 0,
-    "area_city_district": 0,
-    "areaval": 0,
-    "infoformat": 0,
-    "publishdept": 0,
-    "jsondata": 0,
-    "exweigth": 0,
-    "jsoncontent": 0,
-    "sourcehref": 0,
-    "sourcewebsite": 0,
-    "projectname": 0,
-    "bidstatus": 0,
-    "projecthref": 0,
-    "buyer_info": 0,
-    "agency_info": 0,
-    "dataging": 0,
-    "extracttype": 0,
-    "description": 0,
-    "extract_state": 0,
-    "flag_buyer": 0,
-    "check_sensitive": 0,
-    "check_sensitive2": 0,
-    "keywords": 0,
-    "pre_extracttype": 0,
-    "s_sha": 0,
-    "type": 0,
-    "tagname": 0,
-    "kvtext": 0,
-    "repeat": 0,
-    "repeat_id": 0,
-    "repeat_ids": 0,
-    "repeat_reason": 0,
-    "isflow": 0
+  "fusionAllKey": {
+    "String": {
+      "area": "1",
+      "city": "1",
+      "district": "1",
+      "projectcode": "1",
+      "projectscope": "1",
+      "contractcode": "1",
+      "buyer": "1",
+      "buyerperson": "1",
+      "buyertel": "1",
+      "buyeraddr": 1,
+      "agency": "1",
+      "agencyperson": "1",
+      "agencytel": "1",
+      "agencyaddr": "1",
+      "winner": "1",
+      "s_winner": "1",
+      "winnerperson": "1",
+      "winnertel": "1",
+      "winneraddr": "1"
+    },
+    "Int": {
+      "bidopentime": "1"
+    },
+    "Float": {
+      "budget": "1",
+      "bidamount": "1"
+    },
+    "Arr": {
+      "winnerorder": "1"
+    }
   }
 }

udpfusion/src/fusionAddData.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"log"
 	qu "qfw/util"
 	"qfw/util/elastic"
@@ -10,258 +11,9 @@ import (
 )
 
 
-func startTaskAddAddData(data []byte, mapInfo map[string]interface{}) {
-
-	log.Println("开始全量融合流程")
-	defer qu.Catch()
-	//区间id
-	q := map[string]interface{}{
-		"_id": map[string]interface{}{
-			"$gt":  StringTOBsonId(mapInfo["gtid"].(string)),
-			"$lte": StringTOBsonId(mapInfo["lteid"].(string)),
-		},
-	}
-	log.Println("查询条件:",q)
-	sess := mgo.GetMgoConn()
-	defer mgo.DestoryMongoConn(sess)
-	it := sess.DB(mgo.DbName).C(coll_name).Find(&q).Iter()
-	//编译不同的融合组,如何划分组
-	fusionDataGroupMap := make(map[string][]string,0) //待融合组
-
-	norepeatArr,repeatArr,sourceArr,index := make([]string,0),make([]string,0),make([]string,0),0 //重复数据组
-
-	start := int(time.Now().Unix())
-	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
-		if index%10000 == 0 {
-			log.Println("current index",index,tmp["_id"])
-		}
-		tmpId:=BsonTOStringId(tmp["_id"])
-		repeat:=qu.IntAll(tmp["repeat"])
-		sourceid:=qu.ObjToString(tmp["repeat_id"])
-		if repeat==1 {
-			repeatArr = append(repeatArr,tmpId)
-			sourceArr = append(sourceArr,sourceid)
-		}else {
-			norepeatArr = append(norepeatArr,tmpId)
-		}
-
-		tmp = make(map[string]interface{})
-	}
-
-	log.Println("task first:",index,len(norepeatArr),"+",len(repeatArr))
-	log.Println("遍历数据用时：",int(time.Now().Unix())-start,"秒")
-
-	//根据重复组，重新划分新的组别
-	start = int(time.Now().Unix())
-
-	//多线程升索引
-	pool_es := make(chan bool, es_pool)
-	wg_es := &sync.WaitGroup{}
-	tmpEsMap := make(map[string]string,0)
-	isGroupNum := 1000
-	for i:=0;i<len(repeatArr);i++ {
-		if i%10000 == 0 {
-			log.Println("curent index ",i)
-		}
-		if i%isGroupNum==0 && i!=0 {
-			//新的一组执行上一组生索引
-			for k,v:=range tmpEsMap {
-				pool_es <- true
-				wg_es.Add(1)
-				go func(es_id string,cur_ids string) {
-					defer func() {
-						<-pool_es
-						wg_es.Done()
-					}()
-					if es_id!="" && cur_ids!="" {
-						dataArr := *elastic.GetById(esIndex,esType,es_id)
-						if len(dataArr)>0 { //存在-更新
-							allids := qu.ObjToString(dataArr[0]["allids"])
-							allids = allids+","+cur_ids
-							updateStr := `ctx._source.allids=`+ `"`+allids+`"`
-							elastic.Update(esIndex,esType,es_id, updateStr)
-						}else { //不存在-新增
-							savetmp := make(map[string]interface{}, 0)
-							savetmp["allids"] = cur_ids
-							savetmp["_id"] = StringTOBsonId(es_id)
-							savetmp["template_id"] = ""
-							savetmp["fusion_id"] = ""
-							elastic.Save(esIndex, esType, savetmp)
-						}
-					}else {
-						log.Println("异常",es_id,cur_ids)
-					}
-				}(k,v)
-
-			}
-			wg_es.Wait()
-
-			tmpEsMap = make(map[string]string,0)
-		}
-		//新增一条数据
-		repeatid :=repeatArr[i]
-		sourceid := sourceArr[i]
-		if tmpEsMap[sourceid]!="" {
-			ids := tmpEsMap[sourceid]
-			ids = ids+","+repeatid
-			tmpEsMap[sourceid] = ids
-		}else {
-			tmpEsMap[sourceid] = sourceid+","+repeatid
-		}
-	}
-
-	//处理剩余数据
-	if len(tmpEsMap)>0 {
-		for k,v:=range tmpEsMap {
-			pool_es <- true
-			wg_es.Add(1)
-			go func(es_id string,cur_ids string) {
-				defer func() {
-					<-pool_es
-					wg_es.Done()
-				}()
-				dataArr := *elastic.GetById(esIndex,esType,es_id)
-				if len(dataArr)>0 { //存在-更新
-					allids := qu.ObjToString(dataArr[0]["allids"])
-					allids = allids+","+cur_ids
-					updateStr := `ctx._source.allids=`+ `"`+allids+`"`
-					elastic.Update(esIndex,esType,es_id, updateStr)
-				}else { //不存在-新增
-					savetmp := make(map[string]interface{}, 0)
-					savetmp["allids"] = cur_ids
-					savetmp["_id"] = StringTOBsonId(es_id)
-					savetmp["template_id"] = ""
-					savetmp["fusion_id"] = ""
-					elastic.Save(esIndex,esType, savetmp)
-				}
-			}(k,v)
-		}
-		wg_es.Wait()
-		tmpEsMap = make(map[string]string,0)
-
-	}
-
-
-	log.Println("前置索引准备完毕......耗时：",int(time.Now().Unix())-start,"秒")
-
-	start = int(time.Now().Unix())
-	log.Println("开始数据分组... ... ... ...")
-	log.Println("开始数据分组... ... ... ...")
-	log.Println("开始数据分组... ... ... ...")
-
-	//查询分组-多线程
-	for i:=0;i<len(norepeatArr);i++ {
-		if i%10000==0 {
-			log.Println("cur index ",i,norepeatArr[i])
-		}
-		sourceid:=norepeatArr[i]
-		pool_es <- true
-		wg_es.Add(1)
-		go func(sourceid string) {
-			defer func() {
-				<-pool_es
-				wg_es.Done()
-			}()
-			dataArr := *elastic.GetById(esIndex,esType,sourceid)
-			if len(dataArr)>0 { //存在值
-				allids := qu.ObjToString(dataArr[0]["allids"])
-				arr := strings.Split(allids,",")
-				updatelock.Lock()
-				fusionDataGroupMap[sourceid] = arr
-				updatelock.Unlock()
-			}else {
-				arr:=[]string{sourceid}
-				updatelock.Lock()
-				fusionDataGroupMap[sourceid] = arr
-				updatelock.Unlock()
-
-
-			}
-
-		}(sourceid)
-	}
-	wg_es.Wait()
-
-
-	log.Println("最终待融合分组数量：",len(fusionDataGroupMap))
-	log.Println("分组完毕数据用时：",int(time.Now().Unix())-start,"秒")
-	log.Println("********************分割线********************")
-	log.Println("********************分割线********************")
-	log.Println("********************分割线********************")
-
-
-	log.Println("开始进行正式分组融合......先睡秒30秒")
-	time.Sleep(30 * time.Second)
-
-	start = int(time.Now().Unix())
-	//多线程 - 处理数据
-	pool_mgo := make(chan bool, mgo_pool)
-	wg_mgo := &sync.WaitGroup{}
-
-	fusionIndex:=0
-	for k,v:=range fusionDataGroupMap {
-		fusionIndex++
-		pool_mgo <- true
-		wg_mgo.Add(1)
-		go func(sourceid string ,fusionArr []string,fusionIndex int) {
-			defer func() {
-				<-pool_mgo
-				wg_mgo.Done()
-			}()
-			if fusionIndex % 10000==0 {
-				log.Println("数据融合数量：",fusionIndex,sourceid)
-			}
-			weight :=NewWeightData(fusionArr)
-			weight.analyzeBuildStandardData()
-			if len(fusionArr)<=1 { //单组数据-需要新增Es
-				saveFusionData,saveRecordData := weight.dealWithAddFusionStruct()
-				saveid:=mgo.Save(fusion_coll_name,saveFusionData)
-				saveRecordData["_id"] = saveid
-				mgo.Save(record_coll_name,saveRecordData)
-
-				//新增es
-				savetmp := make(map[string]interface{}, 0)
-				fusionid:=BsonTOStringId(saveid)
-				savetmp["_id"] = StringTOBsonId(sourceid)
-				savetmp["allids"] = sourceid
-				savetmp["template_id"] = sourceid
-				savetmp["fusion_id"] = fusionid
-				elastic.Save(esIndex,esType,savetmp)
-
-
-			}else {
-				saveFusionData,saveRecordData := weight.dealWithMultipleAddFusionStruct()
-				saveid:=mgo.Save(fusion_coll_name,saveFusionData)
-				saveRecordData["_id"] = saveid
-				mgo.Save(record_coll_name,saveRecordData)
-
-				//更新数据-融合id-模板id等 `ctx._source.age=101;ctx._source.name="张三"`
-				fusion_id,template_id:=BsonTOStringId(saveid),qu.ObjToString(saveFusionData["fusion_templateid"])
-				updateStr1 := `ctx._source.template_id=`+ `"`+template_id+`";`
-				updateStr2 := `ctx._source.fusion_id=`+ `"`+fusion_id+`"`
-				elastic.Update(esIndex,esType,sourceid, updateStr1+updateStr2)
-
-			}
-		}(k,v,fusionIndex)
-	}
-
-	wg_mgo.Wait()
-
-	log.Println("fusion is over ：",fusionIndex,len(fusionDataGroupMap),"用时：",int(time.Now().Unix())-start,"秒")
-	log.Println("睡眠30秒，然后在发广播")
-
-	time.Sleep(30 * time.Second)
-
-	//任务完成，开始发送广播通知下面节点
-
-	taskSendFusionUdp(mapInfo)
-
-}
-
-
-//增量-融合-一小段
 //func startTaskAddData(data []byte, mapInfo map[string]interface{}) {
-//	log.Println("开始增量融合流程")
+//
+//	log.Println("开始全量融合流程")
 //	defer qu.Catch()
 //	//区间id
 //	q := map[string]interface{}{
@@ -275,14 +27,9 @@ func startTaskAddAddData(data []byte, mapInfo map[string]interface{}) {
 //	defer mgo.DestoryMongoConn(sess)
 //	it := sess.DB(mgo.DbName).C(coll_name).Find(&q).Iter()
 //	//编译不同的融合组,如何划分组
+//	fusionDataGroupMap := make(map[string][]string,0) //待融合组
 //
-//	//待融合组
-//	fusionDataGroupArr := make([][]string,0)
-//	//需要更新组
-//	updateFusionMap,curFusionKeyMap:=make(map[string]interface{},0),make(map[string]interface{},0)
-//	//重复数据组
-//	norepeatArr,repeatArr,sourceArr,index := make([]string,0),make([]string,0),make([]string,0),0
-//
+//	norepeatArr,repeatArr,sourceArr,index := make([]string,0),make([]string,0),make([]string,0),0 //重复数据组
 //
 //	start := int(time.Now().Unix())
 //	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
@@ -296,224 +43,478 @@ func startTaskAddAddData(data []byte, mapInfo map[string]interface{}) {
 //			repeatArr = append(repeatArr,tmpId)
 //			sourceArr = append(sourceArr,sourceid)
 //		}else {
-//			norepeatArr = append(repeatArr,tmpId)
+//			norepeatArr = append(norepeatArr,tmpId)
 //		}
+//
 //		tmp = make(map[string]interface{})
 //	}
 //
-//	log.Println("task first:",index,len(fusionDataGroupArr),"+",len(repeatArr))
+//	log.Println("task first:",index,len(norepeatArr),"+",len(repeatArr))
 //	log.Println("遍历数据用时：",int(time.Now().Unix())-start,"秒")
 //
 //	//根据重复组，重新划分新的组别
 //	start = int(time.Now().Unix())
-//	elastic.InitElasticSize("http://192.168.3.11:9800",10)
+//
+//	//多线程升索引
+//	pool_es := make(chan bool, es_pool)
+//	wg_es := &sync.WaitGroup{}
+//	tmpEsMap := make(map[string]string,0)
+//	isGroupNum := 1000
 //	for i:=0;i<len(repeatArr);i++ {
-//		//查询ES-升索引
-//		repeatid := repeatArr[i]
-//		sourceid := sourceArr[i]
-//		key := fmt.Sprintf("%s",sourceid)
-//		dataArr := *elastic.GetById("allzktest","allzktest",sourceid)
-//		if len(dataArr)>0 { //存在值
+//		if i%10000 == 0 {
+//			log.Println("curent index ",i)
+//		}
+//		if i%isGroupNum==0 && i!=0 {
+//			//新的一组执行上一组生索引
+//			for k,v:=range tmpEsMap {
+//				pool_es <- true
+//				wg_es.Add(1)
+//				go func(es_id string,cur_ids string) {
+//					defer func() {
+//						<-pool_es
+//						wg_es.Done()
+//					}()
+//					if es_id!="" && cur_ids!="" {
+//						dataArr := *elastic.GetById(esIndex,esType,es_id)
+//						if len(dataArr)>0 { //存在-更新
+//							allids := qu.ObjToString(dataArr[0]["allids"])
+//							allids = allids+","+cur_ids
+//							updateStr := `ctx._source.allids=`+ `"`+allids+`"`
+//							elastic.Update(esIndex,esType,es_id, updateStr)
+//						}else { //不存在-新增
+//							savetmp := make(map[string]interface{}, 0)
+//							savetmp["allids"] = cur_ids
+//							savetmp["_id"] = StringTOBsonId(es_id)
+//							savetmp["template_id"] = ""
+//							savetmp["fusion_id"] = ""
+//							elastic.Save(esIndex, esType, savetmp)
+//						}
+//					}else {
+//						log.Println("异常",es_id,cur_ids)
+//					}
+//				}(k,v)
 //
-//			if curFusionKeyMap[key]==nil { //存在融合表-不在当前id段落内
-//				updateFusionMap[key] = ""
-//			}
-//			//es 随时更新ids
-//			allids := qu.ObjToString(dataArr[0]["allids"])
-//			allids = allids+","+repeatid
-//			updateStr := `ctx._source.allids=`+ `"`+allids+`"`
-//			b:=elastic.Update("allzktest","allzktest",sourceid, updateStr)
-//			if !b {
-//				log.Println("es更新异常",repeatid,sourceid)
 //			}
+//			wg_es.Wait()
+//
+//			tmpEsMap = make(map[string]string,0)
+//		}
+//		//新增一条数据
+//		repeatid :=repeatArr[i]
+//		sourceid := sourceArr[i]
+//		if tmpEsMap[sourceid]!="" {
+//			ids := tmpEsMap[sourceid]
+//			ids = ids+","+repeatid
+//			tmpEsMap[sourceid] = ids
 //		}else {
-//			//索引查不到-确定新增- es 随时新增ids
-//			savetmp := make(map[string]interface{}, 0)
-//			savetmp["allids"] = repeatid
-//			savetmp["_id"] = StringTOBsonId(sourceid)
-//			b:=elastic.Save("allzktest", "allzktest", savetmp)
-//			if !b {
-//				log.Println("es保存异常",repeatid,sourceid)
-//			}
-//			curFusionKeyMap[key] = ""
+//			tmpEsMap[sourceid] = sourceid+","+repeatid
 //		}
 //	}
 //
-//	log.Println("前置索引准备完毕... ...","耗时：",int(time.Now().Unix())-start,"秒")
+//	//处理剩余数据
+//	if len(tmpEsMap)>0 {
+//		for k,v:=range tmpEsMap {
+//			pool_es <- true
+//			wg_es.Add(1)
+//			go func(es_id string,cur_ids string) {
+//				defer func() {
+//					<-pool_es
+//					wg_es.Done()
+//				}()
+//				dataArr := *elastic.GetById(esIndex,esType,es_id)
+//				if len(dataArr)>0 { //存在-更新
+//					allids := qu.ObjToString(dataArr[0]["allids"])
+//					allids = allids+","+cur_ids
+//					updateStr := `ctx._source.allids=`+ `"`+allids+`"`
+//					elastic.Update(esIndex,esType,es_id, updateStr)
+//				}else { //不存在-新增
+//					savetmp := make(map[string]interface{}, 0)
+//					savetmp["allids"] = cur_ids
+//					savetmp["_id"] = StringTOBsonId(es_id)
+//					savetmp["template_id"] = ""
+//					savetmp["fusion_id"] = ""
+//					elastic.Save(esIndex,esType, savetmp)
+//				}
+//			}(k,v)
+//		}
+//		wg_es.Wait()
+//		tmpEsMap = make(map[string]string,0)
+//
+//	}
 //
 //
+//	log.Println("前置索引准备完毕......耗时：",int(time.Now().Unix())-start,"秒")
+//
 //	start = int(time.Now().Unix())
 //	log.Println("开始数据分组... ... ... ...")
 //	log.Println("开始数据分组... ... ... ...")
 //	log.Println("开始数据分组... ... ... ...")
 //
-//	//当前段落组
+//	//查询分组-多线程
 //	for i:=0;i<len(norepeatArr);i++ {
-//		sourceid:=norepeatArr[i]
-//		dataArr := *elastic.GetById("allzktest","allzktest",sourceid)
-//		if len(dataArr)>0 { //存在值
-//			allids := qu.ObjToString(dataArr[0]["allids"])
-//			arr := strings.Split(allids,",")
-//			arr = append(arr,sourceid)
-//			fusionDataGroupArr = append(fusionDataGroupArr,arr)
-//		}else {
-//			arr:=[]string{sourceid}
-//			fusionDataGroupArr = append(fusionDataGroupArr,arr)
-//		}
-//	}
-//	//更新组
-//	for k,_:=range updateFusionMap {
-//		sourceid:=qu.ObjToString(k)
-//		dataArr := *elastic.GetById("allzktest","allzktest",sourceid)
-//		if len(dataArr)>0 { //存在值
-//			allids := qu.ObjToString(dataArr[0]["allids"])
-//			arr := strings.Split(allids,",")
-//			arr = append(arr,sourceid)
-//			fusionDataGroupArr = append(fusionDataGroupArr,arr)
-//		}else {
-//			log.Println("融合表更新，查询Es异常：",sourceid)
+//		if i%10000==0 {
+//			log.Println("cur index ",i,norepeatArr[i])
 //		}
-//	}
+//		sourceid:=norepeatArr[i]
+//		pool_es <- true
+//		wg_es.Add(1)
+//		go func(sourceid string) {
+//			defer func() {
+//				<-pool_es
+//				wg_es.Done()
+//			}()
+//			dataArr := *elastic.GetById(esIndex,esType,sourceid)
+//			if len(dataArr)>0 { //存在值
+//				allids := qu.ObjToString(dataArr[0]["allids"])
+//				arr := strings.Split(allids,",")
+//				updatelock.Lock()
+//				fusionDataGroupMap[sourceid] = arr
+//				updatelock.Unlock()
+//			}else {
+//				arr:=[]string{sourceid}
+//				updatelock.Lock()
+//				fusionDataGroupMap[sourceid] = arr
+//				updatelock.Unlock()
+//
 //
+//			}
+//
+//		}(sourceid)
+//	}
+//	wg_es.Wait()
 //
 //
-//	//isErrNum:=0
-//	//for i:=0;i<len(repeatArr);i++ {
-//	//	sourceid := sourceArr[i]
-//	//	isAddExist,index := false,0
-//	//	//根据原sourceid 直接遍历组
-//	//R:	for k,v:=range fusionDataGroupArr{
-//	//		for _,v1:=range v{
-//	//			if v1==sourceid {
-//	//				index = k
-//	//				isAddExist = true
-//	//				break R
-//	//			}
-//	//		}
-//	//	}
-//	//	if i%1000 == 0 {
-//	//		log.Println("分组中...","current index",i,repeatArr[i])
-//	//	}
-//	//
-//	//	if isAddExist { //数组截取替换-找到指定
-//	//		arr := make([]string,0)
-//	//		arr = fusionDataGroupArr[index]
-//	//		arr = append(arr,repeatArr[i])//组拼接当前id
-//	//		fusionDataGroupArr[index] = arr
-//	//		log.Println("... ... 正常单组新增",i)
-//	//
-//	//	}else {//当前段落未找到-需要查询融合表，,遍历融合表
-//	//		arr,fusionTmpData := make([]string,0),make(map[string]interface{},0)
-//	//		arr,fusionTmpData = dealWithFindFusionDataArr(sourceid)
-//	//		arr = append(arr,repeatArr[i])//组拼接当前id
-//	//
-//	//
-//	//
-//	//
-//	//
-//	//		if len(arr)==1 { //异常错误，新增
-//	//			isErrNum++
-//	//			log.Println("... ... 数据异常异常,融合表，当前组均找不到数据",repeatArr[i])
-//	//			arr_error := make([]string,0)
-//	//			arr_error = append(arr_error,repeatArr[i])//组拼接当前id
-//	//			fusionDataGroupArr = append(fusionDataGroupArr,arr_error)
-//	//			addOrUpdateArr = append(addOrUpdateArr,false)
-//	//			infoFusionArr = append(infoFusionArr, map[string]interface{}{})
-//	//		}else { //正常更新
-//	//			log.Println("... ... 正常多组新增",i)
-//	//			fusionDataGroupArr = append(fusionDataGroupArr,arr)
-//	//			addOrUpdateArr = append(addOrUpdateArr,true)
-//	//			infoFusionArr = append(infoFusionArr,fusionTmpData)
-//	//		}
-//	//
-//	//	}
-//	//	//不断改变中
-//	//	if i%1000 == 0 {
-//	//		log.Println("当前分组数量：",len(fusionDataGroupArr))
-//	//	}
-//	//}
-//
-//
-//
-//
-//	log.Println("最终待融合分组数量：",len(fusionDataGroupArr))
+//	log.Println("最终待融合分组数量：",len(fusionDataGroupMap))
 //	log.Println("分组完毕数据用时：",int(time.Now().Unix())-start,"秒")
 //	log.Println("********************分割线********************")
 //	log.Println("********************分割线********************")
 //	log.Println("********************分割线********************")
 //
 //
-//	log.Println("开始处理分组融合... ... ... ...")
-//	log.Println("开始处理分组融合... ... ... ...")
-//	log.Println("开始处理分组融合... ... ... ...")
+//	log.Println("开始进行正式分组融合......先睡秒30秒")
+//	time.Sleep(30 * time.Second)
 //
 //	start = int(time.Now().Unix())
 //	//多线程 - 处理数据
-//	pool := make(chan bool, 3)
-//	wg := &sync.WaitGroup{}
-//
-//	for i:=0;i<len(fusionDataGroupArr);i++ {
-//		fusionArr := fusionDataGroupArr[i]
-//		pool <- true
-//		wg.Add(1)
-//		go func(fusionArr []string,i int) {
+//	pool_mgo := make(chan bool, mgo_pool)
+//	wg_mgo := &sync.WaitGroup{}
+//
+//	fusionIndex:=0
+//	for k,v:=range fusionDataGroupMap {
+//		fusionIndex++
+//		pool_mgo <- true
+//		wg_mgo.Add(1)
+//		go func(sourceid string ,fusionArr []string,fusionIndex int) {
 //			defer func() {
-//				<-pool
-//				wg.Done()
+//				<-pool_mgo
+//				wg_mgo.Done()
 //			}()
-//			//构建数据
-//			if (i+1)%500 == 0 {
-//				log.Println("构建第",i+1,"组数据...","数量：",len(fusionArr),fusionArr)
+//			if fusionIndex % 10000==0 {
+//				log.Println("数据融合数量：",fusionIndex,sourceid)
 //			}
 //			weight :=NewWeightData(fusionArr)
-//			////整理数据-筛选排名，模板
 //			weight.analyzeBuildStandardData()
-//
-//			if len(fusionArr)<=1 {
+//			if len(fusionArr)<=1 { //单组数据-需要新增Es
 //				saveFusionData,saveRecordData := weight.dealWithAddFusionStruct()
 //				saveid:=mgo.Save(fusion_coll_name,saveFusionData)
 //				saveRecordData["_id"] = saveid
 //				mgo.Save(record_coll_name,saveRecordData)
-//			}else {
-//				//if addOrUpdateArr[i] {
-//				//	//log.Println("多组更新... ...")
-//				//	tmpdata:=infoFusionArr[i]
-//				//	updateFusionData,updateRecordData := weight.dealWithMultipleUpdateFusionStruct(tmpdata)
-//				//
-//				//	UpdateFusion.updatePool <- []map[string]interface{}{
-//				//		map[string]interface{}{
-//				//			"_id": tmpdata["_id"],
-//				//		},
-//				//		updateFusionData,
-//				//	}
-//				//	UpdateRecord.updatePool <- []map[string]interface{}{
-//				//		map[string]interface{}{
-//				//			"_id": tmpdata["_id"],
-//				//		},
-//				//		updateRecordData,
-//				//	}
-//				//}else {
-//				//	//log.Println("多组生成... ...")
-//				//	saveFusionData,saveRecordData := weight.dealWithMultipleAddFusionStruct()
-//				//	saveid:=mgo.Save(fusion_coll_name,saveFusionData)
-//				//	saveRecordData["_id"] = saveid
-//				//	mgo.Save(record_coll_name,saveRecordData)
-//				//}
-//			}
 //
+//				//新增es
+//				savetmp := make(map[string]interface{}, 0)
+//				fusionid:=BsonTOStringId(saveid)
+//				savetmp["_id"] = StringTOBsonId(sourceid)
+//				savetmp["allids"] = sourceid
+//				savetmp["template_id"] = sourceid
+//				savetmp["fusion_id"] = fusionid
+//				elastic.Save(esIndex,esType,savetmp)
 //
-//		}(fusionArr,i)
 //
+//			}else {
+//				saveFusionData,saveRecordData := weight.dealWithMultipleAddFusionStruct()
+//				saveid:=mgo.Save(fusion_coll_name,saveFusionData)
+//				saveRecordData["_id"] = saveid
+//				mgo.Save(record_coll_name,saveRecordData)
 //
+//				//更新数据-融合id-模板id等 `ctx._source.age=101;ctx._source.name="张三"`
+//				fusion_id,template_id:=BsonTOStringId(saveid),qu.ObjToString(saveFusionData["fusion_templateid"])
+//				updateStr1 := `ctx._source.template_id=`+ `"`+template_id+`";`
+//				updateStr2 := `ctx._source.fusion_id=`+ `"`+fusion_id+`"`
+//				elastic.Update(esIndex,esType,sourceid, updateStr1+updateStr2)
 //
+//			}
+//		}(k,v,fusionIndex)
 //	}
 //
-//	wg.Wait()
+//	wg_mgo.Wait()
 //
-//	log.Println("fusion is over ：",len(fusionDataGroupArr),"用时：",int(time.Now().Unix())-start,"秒")
+//	log.Println("fusion is over ：",fusionIndex,len(fusionDataGroupMap),"用时：",int(time.Now().Unix())-start,"秒")
 //	log.Println("睡眠30秒，然后在发广播")
+//
 //	time.Sleep(30 * time.Second)
+//
 //	//任务完成，开始发送广播通知下面节点
+//
 //	taskSendFusionUdp(mapInfo)
 //
-//}
+//}
+
+
+//增量-融合一小段
+func startTaskAddData(data []byte, mapInfo map[string]interface{}) {
+	log.Println("开始增量融合流程")
+	defer qu.Catch()
+	//区间id
+	q := map[string]interface{}{
+		"_id": map[string]interface{}{
+			"$gt":  StringTOBsonId(mapInfo["gtid"].(string)),
+			"$lte": StringTOBsonId(mapInfo["lteid"].(string)),
+		},
+	}
+	log.Println("查询条件:",q)
+	sess := mgo.GetMgoConn()
+	defer mgo.DestoryMongoConn(sess)
+	it := sess.DB(mgo.DbName).C(coll_name).Find(&q).Iter()
+	//编译不同的融合组,如何划分组
+
+	//待融合组
+	fusionDataGroupArr := make([][]string,0)
+	//需要更新组
+	updateFusionMap,curFusionKeyMap:=make(map[string]interface{},0),make(map[string]interface{},0)
+	//重复数据组
+	norepeatArr,repeatArr,sourceArr,index := make([]string,0),make([]string,0),make([]string,0),0
+
+
+	start := int(time.Now().Unix())
+	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
+		if index%10000 == 0 {
+			log.Println("current index",index,tmp["_id"])
+		}
+		tmpId:=BsonTOStringId(tmp["_id"])
+		repeat:=qu.IntAll(tmp["repeat"])
+		sourceid:=qu.ObjToString(tmp["repeat_id"])
+		if repeat==1 {
+			repeatArr = append(repeatArr,tmpId)
+			sourceArr = append(sourceArr,sourceid)
+		}else {
+			norepeatArr = append(repeatArr,tmpId)
+		}
+		tmp = make(map[string]interface{})
+	}
+
+	log.Println("task first:",index,len(fusionDataGroupArr),"+",len(repeatArr))
+	log.Println("遍历数据用时：",int(time.Now().Unix())-start,"秒")
+
+	//根据重复组，重新划分新的组别
+	start = int(time.Now().Unix())
+	elastic.InitElasticSize("http://192.168.3.11:9800",10)
+	for i:=0;i<len(repeatArr);i++ {
+		//查询ES-升索引
+		repeatid := repeatArr[i]
+		sourceid := sourceArr[i]
+		key := fmt.Sprintf("%s",sourceid)
+		dataArr := *elastic.GetById("allzktest","allzktest",sourceid)
+		if len(dataArr)>0 { //存在值
+
+			if curFusionKeyMap[key]==nil { //存在融合表-不在当前id段落内
+				updateFusionMap[key] = ""
+			}
+			//es 随时更新ids
+			allids := qu.ObjToString(dataArr[0]["allids"])
+			allids = allids+","+repeatid
+			updateStr := `ctx._source.allids=`+ `"`+allids+`"`
+			b:=elastic.Update("allzktest","allzktest",sourceid, updateStr)
+			if !b {
+				log.Println("es更新异常",repeatid,sourceid)
+			}
+		}else {
+			//索引查不到-确定新增- es 随时新增ids
+			savetmp := make(map[string]interface{}, 0)
+			savetmp["allids"] = repeatid
+			savetmp["_id"] = StringTOBsonId(sourceid)
+			b:=elastic.Save("allzktest", "allzktest", savetmp)
+			if !b {
+				log.Println("es保存异常",repeatid,sourceid)
+			}
+			curFusionKeyMap[key] = ""
+		}
+	}
+
+	log.Println("前置索引准备完毕... ...","耗时：",int(time.Now().Unix())-start,"秒")
+
+
+	start = int(time.Now().Unix())
+	log.Println("开始数据分组... ... ... ...")
+	log.Println("开始数据分组... ... ... ...")
+	log.Println("开始数据分组... ... ... ...")
+
+	//当前段落组
+	for i:=0;i<len(norepeatArr);i++ {
+		sourceid:=norepeatArr[i]
+		dataArr := *elastic.GetById("allzktest","allzktest",sourceid)
+		if len(dataArr)>0 { //存在值
+			allids := qu.ObjToString(dataArr[0]["allids"])
+			arr := strings.Split(allids,",")
+			arr = append(arr,sourceid)
+			fusionDataGroupArr = append(fusionDataGroupArr,arr)
+		}else {
+			arr:=[]string{sourceid}
+			fusionDataGroupArr = append(fusionDataGroupArr,arr)
+		}
+	}
+	//更新组
+	for k,_:=range updateFusionMap {
+		sourceid:=qu.ObjToString(k)
+		dataArr := *elastic.GetById("allzktest","allzktest",sourceid)
+		if len(dataArr)>0 { //存在值
+			allids := qu.ObjToString(dataArr[0]["allids"])
+			arr := strings.Split(allids,",")
+			arr = append(arr,sourceid)
+			fusionDataGroupArr = append(fusionDataGroupArr,arr)
+		}else {
+			log.Println("融合表更新，查询Es异常：",sourceid)
+		}
+	}
+
+
+
+	//isErrNum:=0
+	//for i:=0;i<len(repeatArr);i++ {
+	//	sourceid := sourceArr[i]
+	//	isAddExist,index := false,0
+	//	//根据原sourceid 直接遍历组
+	//R:	for k,v:=range fusionDataGroupArr{
+	//		for _,v1:=range v{
+	//			if v1==sourceid {
+	//				index = k
+	//				isAddExist = true
+	//				break R
+	//			}
+	//		}
+	//	}
+	//	if i%1000 == 0 {
+	//		log.Println("分组中...","current index",i,repeatArr[i])
+	//	}
+	//
+	//	if isAddExist { //数组截取替换-找到指定
+	//		arr := make([]string,0)
+	//		arr = fusionDataGroupArr[index]
+	//		arr = append(arr,repeatArr[i])//组拼接当前id
+	//		fusionDataGroupArr[index] = arr
+	//		log.Println("... ... 正常单组新增",i)
+	//
+	//	}else {//当前段落未找到-需要查询融合表，,遍历融合表
+	//		arr,fusionTmpData := make([]string,0),make(map[string]interface{},0)
+	//		arr,fusionTmpData = dealWithFindFusionDataArr(sourceid)
+	//		arr = append(arr,repeatArr[i])//组拼接当前id
+	//
+	//
+	//
+	//
+	//
+	//		if len(arr)==1 { //异常错误，新增
+	//			isErrNum++
+	//			log.Println("... ... 数据异常异常,融合表，当前组均找不到数据",repeatArr[i])
+	//			arr_error := make([]string,0)
+	//			arr_error = append(arr_error,repeatArr[i])//组拼接当前id
+	//			fusionDataGroupArr = append(fusionDataGroupArr,arr_error)
+	//			addOrUpdateArr = append(addOrUpdateArr,false)
+	//			infoFusionArr = append(infoFusionArr, map[string]interface{}{})
+	//		}else { //正常更新
+	//			log.Println("... ... 正常多组新增",i)
+	//			fusionDataGroupArr = append(fusionDataGroupArr,arr)
+	//			addOrUpdateArr = append(addOrUpdateArr,true)
+	//			infoFusionArr = append(infoFusionArr,fusionTmpData)
+	//		}
+	//
+	//	}
+	//	//不断改变中
+	//	if i%1000 == 0 {
+	//		log.Println("当前分组数量：",len(fusionDataGroupArr))
+	//	}
+	//}
+
+
+
+
+	log.Println("最终待融合分组数量：",len(fusionDataGroupArr))
+	log.Println("分组完毕数据用时：",int(time.Now().Unix())-start,"秒")
+	log.Println("********************分割线********************")
+	log.Println("********************分割线********************")
+	log.Println("********************分割线********************")
+
+
+	log.Println("开始处理分组融合... ... ... ...")
+	log.Println("开始处理分组融合... ... ... ...")
+	log.Println("开始处理分组融合... ... ... ...")
+
+	start = int(time.Now().Unix())
+	//多线程 - 处理数据
+	pool := make(chan bool, 3)
+	wg := &sync.WaitGroup{}
+
+	for i:=0;i<len(fusionDataGroupArr);i++ {
+		fusionArr := fusionDataGroupArr[i]
+		pool <- true
+		wg.Add(1)
+		go func(fusionArr []string,i int) {
+			defer func() {
+				<-pool
+				wg.Done()
+			}()
+			//构建数据
+			if (i+1)%500 == 0 {
+				log.Println("构建第",i+1,"组数据...","数量：",len(fusionArr),fusionArr)
+			}
+			weight :=NewWeightData(fusionArr)
+			////整理数据-筛选排名，模板
+			weight.analyzeBuildStandardData()
+
+			if len(fusionArr)<=1 {
+				saveFusionData,saveRecordData := weight.dealWithAddFusionStruct()
+				saveid:=mgo.Save(fusion_coll_name,saveFusionData)
+				saveRecordData["_id"] = saveid
+				mgo.Save(record_coll_name,saveRecordData)
+			}else {
+				//if addOrUpdateArr[i] {
+				//	//log.Println("多组更新... ...")
+				//	tmpdata:=infoFusionArr[i]
+				//	updateFusionData,updateRecordData := weight.dealWithMultipleUpdateFusionStruct(tmpdata)
+				//
+				//	UpdateFusion.updatePool <- []map[string]interface{}{
+				//		map[string]interface{}{
+				//			"_id": tmpdata["_id"],
+				//		},
+				//		updateFusionData,
+				//	}
+				//	UpdateRecord.updatePool <- []map[string]interface{}{
+				//		map[string]interface{}{
+				//			"_id": tmpdata["_id"],
+				//		},
+				//		updateRecordData,
+				//	}
+				//}else {
+				//	//log.Println("多组生成... ...")
+				//	saveFusionData,saveRecordData := weight.dealWithMultipleAddFusionStruct()
+				//	saveid:=mgo.Save(fusion_coll_name,saveFusionData)
+				//	saveRecordData["_id"] = saveid
+				//	mgo.Save(record_coll_name,saveRecordData)
+				//}
+			}
+
+
+		}(fusionArr,i)
+
+
+
+	}
+
+	wg.Wait()
+
+	log.Println("fusion is over ：",len(fusionDataGroupArr),"用时：",int(time.Now().Unix())-start,"秒")
+	log.Println("睡眠30秒，然后在发广播")
+	time.Sleep(30 * time.Second)
+	//任务完成，开始发送广播通知下面节点
+	taskSendFusionUdp(mapInfo)
+
+}

udpfusion/src/fusionFullData.go

@@ -15,11 +15,11 @@ import (
 func startTaskFullData(data []byte, mapInfo map[string]interface{}) {
 
 	//临时测试
-	//先到处具体需要融合组数据-存mongo
-	exportFusionMongoData()
-	time.Sleep(60 * time.Second)
-
-	//具体融合数据的方法
+	//先导出具体需要融合组数据组-存mongo
+	//exportFusionMongoData()
+	//time.Sleep(60 * time.Second)
+	//
+	////具体融合数据的方法
 	startFusionData()
 	time.Sleep(60 * time.Second)
 	return
@@ -69,8 +69,6 @@ func startTaskFullData(data []byte, mapInfo map[string]interface{}) {
 							savetmp := make(map[string]interface{}, 0)
 							savetmp["allids"] = cur_ids
 							savetmp["_id"] = StringTOBsonId(es_id)
-							savetmp["template_id"] = ""
-							savetmp["fusion_id"] = ""
 							elastic.Save(esIndex, esType, savetmp)
 						}
 					}else {
@@ -126,8 +124,6 @@ func startTaskFullData(data []byte, mapInfo map[string]interface{}) {
 						savetmp := make(map[string]interface{}, 0)
 						savetmp["allids"] = cur_ids
 						savetmp["_id"] = StringTOBsonId(es_id)
-						savetmp["template_id"] = ""
-						savetmp["fusion_id"] = ""
 						elastic.Save(esIndex, esType, savetmp)
 					}
 				}else {
@@ -149,7 +145,7 @@ func startTaskFullData(data []byte, mapInfo map[string]interface{}) {
 	startFusionData()
 	time.Sleep(60 * time.Second)
 
-	taskSendFusionUdp(mapInfo)
+	log.Println("all fusion is over ... ...")
 }
 
 

udpfusion/src/main.go

@@ -21,7 +21,7 @@ var (
 	coll_name 	 						string
 	fusion_coll_name,record_coll_name	string						//新增表名
 	group_coll_name						string						//融合组表记录
-	NoNeedFusionKey 					map[string]interface{}   	//不需要融合的key
+	fusionAllKey 						map[string]interface{}   	//不需要融合的key
 	siteJsonData						map[string]string			//站点池
 	esIndex,esType					    string						//索引-类型
 	mgo_pool,es_pool,isgroupfn			int
@@ -54,7 +54,7 @@ func initMgoAndSite()  {
 	fusion_coll_name = sysconfig["fusion_coll_name"].(string)
 	record_coll_name = sysconfig["record_coll_name"].(string)
 	isgroupfn = qu.IntAllDef(sysconfig["isgroupfn"], 10000)
-	NoNeedFusionKey = sysconfig["notFusionKey"].(map[string]interface{})
+	fusionAllKey = sysconfig["fusionAllKey"].(map[string]interface{})
 
 
 	site := mgoconf["site"].(map[string]interface{})
@@ -179,7 +179,7 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 //结束发送udp
 func taskSendFusionUdp(mapinfo map[string]interface{})  {
 
-	//log.Println("信息融合结束-发送udp")
+	log.Println("信息融合结束-发送udp")
 	for _, to := range nextNode {
 		sid, _ := mapinfo["gtid"].(string)
 		eid, _ := mapinfo["lteid"].(string)
@@ -198,6 +198,7 @@ func taskSendFusionUdp(mapinfo map[string]interface{})  {
 		udptaskmap.Store(key, node)
 		udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
 	}
+
 }
 
 

udpfusion/src/mark

@@ -0,0 +1,426 @@
+package main
+
+import (
+	"encoding/json"
+	"log"
+	qu "qfw/util"
+	"qfw/util/elastic"
+	"strings"
+	"sync"
+	"time"
+	es_elastic "qfw/common/src/gopkg.in/olivere/elastic.v1"
+)
+
+
+func startTaskFullData(data []byte, mapInfo map[string]interface{}) {
+
+	//临时测试
+	//先导出具体需要融合组数据组-存mongo
+	exportFusionMongoData()
+	time.Sleep(60 * time.Second)
+
+	//具体融合数据的方法
+	startFusionData()
+	time.Sleep(60 * time.Second)
+	return
+
+
+	log.Println("开始全量融合流程")
+	defer qu.Catch()
+	//区间id
+	q := map[string]interface{}{
+		"_id": map[string]interface{}{
+			"$gt":  StringTOBsonId(mapInfo["gtid"].(string)),
+			"$lte": StringTOBsonId(mapInfo["lteid"].(string)),
+		},
+	}
+	log.Println("查询条件:",q)
+	sess := mgo.GetMgoConn()
+	defer mgo.DestoryMongoConn(sess)
+	it := sess.DB(mgo.DbName).C(coll_name).Find(&q).Iter()
+	index,isOK:=0,0
+	start := int(time.Now().Unix())
+	//多线程升索引
+	pool_es := make(chan bool, es_pool)
+	wg_es := &sync.WaitGroup{}
+	tmpEsMap := make(map[string]string,0)
+
+	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
+		//每遍历isgroupfn条 划分组别
+		if index%isgroupfn==0 && index!=0 {
+			log.Println("current index",index,tmp["_id"])
+			//新的一组执行上一组生索引
+			for k,v:=range tmpEsMap {
+				pool_es <- true
+				wg_es.Add(1)
+				go func(es_id string,cur_ids string) {
+					defer func() {
+						<-pool_es
+						wg_es.Done()
+					}()
+					if es_id!="" && cur_ids!="" {
+						dataArr := *elastic.GetById(esIndex,esType,es_id)
+						if len(dataArr)>0 { //存在-更新
+							allids := qu.ObjToString(dataArr[0]["allids"])
+							allids = allids+","+cur_ids
+							updateStr := `ctx._source.allids=`+ `"`+allids+`"`
+							elastic.Update(esIndex,esType,es_id, updateStr)
+						}else { //不存在-新增
+							savetmp := make(map[string]interface{}, 0)
+							savetmp["allids"] = cur_ids
+							savetmp["_id"] = StringTOBsonId(es_id)
+							savetmp["template_id"] = ""
+							savetmp["fusion_id"] = ""
+							elastic.Save(esIndex, esType, savetmp)
+						}
+					}else {
+						log.Println("异常",es_id,cur_ids)
+					}
+				}(k,v)
+
+			}
+			wg_es.Wait()
+			tmpEsMap = make(map[string]string,0)
+		}
+
+		repeat := qu.IntAll(tmp["repeat"])
+		sourceid := BsonTOStringId(tmp["_id"])
+		repeatid := BsonTOStringId(tmp["_id"])
+		if repeat==1 {
+			sourceid = qu.ObjToString(tmp["repeat_id"])
+		}else {
+			isOK++
+		}
+		if tmpEsMap[sourceid]!="" {
+			ids := tmpEsMap[sourceid]
+			ids = ids+","+repeatid
+			tmpEsMap[sourceid] = ids
+		}else {
+			tmpEsMap[sourceid] = repeatid
+		}
+
+		tmp = make(map[string]interface{})
+	}
+
+	log.Println("task first:",index,"不重复数：",isOK,"遍历分组数据用时：",int(time.Now().Unix())-start,"秒")
+
+	//处理剩余数据
+	if len(tmpEsMap)>0 {
+		log.Println("处理剩余数据：",len(tmpEsMap))
+		for k,v:=range tmpEsMap {
+			pool_es <- true
+			wg_es.Add(1)
+			go func(es_id string,cur_ids string) {
+				defer func() {
+					<-pool_es
+					wg_es.Done()
+				}()
+				if es_id!="" && cur_ids!="" {
+					dataArr := *elastic.GetById(esIndex,esType,es_id)
+					if len(dataArr)>0 { //存在-更新
+						allids := qu.ObjToString(dataArr[0]["allids"])
+						allids = allids+","+cur_ids
+						updateStr := `ctx._source.allids=`+ `"`+allids+`"`
+						elastic.Update(esIndex,esType,es_id, updateStr)
+					}else { //不存在-新增
+						savetmp := make(map[string]interface{}, 0)
+						savetmp["allids"] = cur_ids
+						savetmp["_id"] = StringTOBsonId(es_id)
+						savetmp["template_id"] = ""
+						savetmp["fusion_id"] = ""
+						elastic.Save(esIndex, esType, savetmp)
+					}
+				}else {
+					log.Println("异常",es_id,cur_ids)
+				}
+			}(k,v)
+		}
+		wg_es.Wait()
+		tmpEsMap = make(map[string]string,0)
+
+	}
+	log.Println("索引准备完毕睡眠30s......耗时：",int(time.Now().Unix())-start,"秒")
+	time.Sleep(60 * time.Second)
+
+	//先到处具体需要融合组数据-存mongo
+	exportFusionMongoData()
+	time.Sleep(60 * time.Second)
+	//具体融合数据的方法
+	startFusionData()
+	time.Sleep(60 * time.Second)
+
+	taskSendFusionUdp(mapInfo)
+}
+
+
+
+
+func exportFusionMongoData()  {
+
+	start := int(time.Now().Unix())
+	log.Println("开始导出融合组数据......")
+	//遍历索引
+	esclient := elastic.GetEsConn()
+	defer elastic.DestoryEsConn(esclient)
+
+	if esclient == nil {
+		log.Println("连接池异常")
+	}
+	q :=es_elastic.NewBoolQuery()
+	cursor, err := esclient.Scan(esIndex).Query(es_elastic.NewBoolQuery().Must(q)).
+		Size(200).Do()
+	if err != nil {
+		log.Println("cursor",err)
+	}
+	if cursor.Results == nil {
+		log.Println("results != nil; got nil")
+	}
+	if cursor.Results.Hits == nil {
+		log.Println("expected results.Hits != nil; got nil")
+	}
+	log.Println("查询正常-总数：",cursor.TotalHits())
+	//多线程 - 处理数据
+	pool_es := make(chan bool, es_pool)
+	wg_es := &sync.WaitGroup{}
+	pages,numDocs := 0,0
+	for {
+		searchResult, err := cursor.Next()
+		if err != nil {
+			if err.Error() == "EOS" {
+				break
+			}else {
+				log.Println("cursor searchResult",err)
+			}
+		}
+		pages++
+		isLog := false
+		for _, hit := range searchResult.Hits.Hits {
+			tmp := make(map[string]interface{})
+			err := json.Unmarshal(*hit.Source, &tmp)
+			if err != nil {
+				log.Println("json Unmarshal error")
+				continue
+			}
+			if !isLog && numDocs%10000==0 {
+				log.Println("当前条数：", numDocs, "Es数据：", tmp["_id"])
+				isLog = true
+			}
+
+			numDocs++
+			fusion_ids := qu.ObjToString(tmp["allids"])
+			sourceid := qu.ObjToString(tmp["_id"])
+			pool_es <- true
+			wg_es.Add(1)
+			go func(sourceid string, fusionArr string) {
+				defer func() {
+					<-pool_es
+					wg_es.Done()
+				}()
+				AddGroupPool.pool <- map[string]interface{}{
+					"_id":StringTOBsonId(sourceid),
+					"allids":fusion_ids,
+				}
+			}(sourceid, fusion_ids)
+		}
+
+	}
+	log.Println("遍历Es结束......")
+	wg_es.Wait()
+
+	log.Println("fusion group over ：",numDocs,"用时：",int(time.Now().Unix())-start,"秒")
+}
+
+func startFusionData()  {
+	log.Println("开始全量融合流程...")
+	defer qu.Catch()
+	//可以开多程序-不同id段执行融合
+	q := map[string]interface{}{}
+	sess := mgo.GetMgoConn()
+	defer mgo.DestoryMongoConn(sess)
+	it := sess.DB(mgo.DbName).C(group_coll_name).Find(&q).Iter()
+	index,start :=0, int(time.Now().Unix())
+	//多线程保存数据
+	pool_mgo := make(chan bool, mgo_pool)
+	wg_mgo := &sync.WaitGroup{}
+	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
+		if index%10000==0 {
+			log.Println("current index",index,tmp["_id"])
+		}
+		fusion_ids := qu.ObjToString(tmp["allids"])
+		fusionArr := strings.Split(fusion_ids, ",")
+		sourceid := BsonTOStringId(tmp["_id"])
+		pool_mgo <- true
+		wg_mgo.Add(1)
+		go func(sourceid string, fusionArr []string) {
+			defer func() {
+				<-pool_mgo
+				wg_mgo.Done()
+			}()
+			weight := NewWeightData(fusionArr)
+			weight.analyzeBuildStandardData()
+
+			saveFusionData, saveRecordData:= map[string]interface{}{},map[string]interface{}{}
+			if len(fusionArr) <= 1 {
+				saveFusionData, saveRecordData = weight.dealWithAddFusionStruct()
+			}else {
+				saveFusionData, saveRecordData = weight.dealWithMultipleAddFusionStruct()
+			}
+			//新增融合表
+			saveid := mgo.Save(fusion_coll_name, saveFusionData)
+			saveRecordData["_id"] = saveid
+			//批量新增日志表
+			AddRecordPool.pool <- saveRecordData
+			//批量更新分组表
+			UpdateGroupPool.pool <- []map[string]interface{}{
+				map[string]interface{}{
+					"_id": StringTOBsonId(sourceid),
+				},
+				map[string]interface{}{
+					"$set": map[string]interface{}{
+						"fusion_id": BsonTOStringId(saveid),
+						"template_id":qu.ObjToString(saveFusionData["fusion_templateid"]),
+					},
+				},
+			}
+		}(sourceid, fusionArr)
+
+		tmp = make(map[string]interface{})
+	}
+	wg_mgo.Wait()
+
+	log.Println("fusion is over:",index,"总用时：",int(time.Now().Unix())-start,"秒")
+
+}
+
+
+//处理结构数据
+func (weight *weightDataMap)dealWithStructData(recordDict *map[string]interface{}) map[string]interface{} {
+
+	//模板id 数据
+	templateid:=weight.templateid
+	templateTmp:=weight.data[templateid].data
+
+	modifyData :=make(map[string]interface{},0)
+
+	attach_text,isAttach:=make(map[string]interface{},0),false
+	if tmp_arr,b := templateTmp["attach_text"].(map[string]interface{});b {
+		//有值符合-
+		attach_text = tmp_arr
+	}
+	//附件判重-并合并新增
+	keyIndex := -1
+	for k,_:=range attach_text {
+		key:=qu.IntAll(k)
+		if key>keyIndex {
+			keyIndex = key
+		}
+	}
+	for _,value_id :=range weight.saveids {
+		if templateid == value_id {
+			continue
+		}
+		rankData := weight.data[value_id].data //具体其他排名数据
+		if attachData,b := rankData["attach_text"].(map[string]interface{});b {
+			if len(attachData)>0  { //有值
+				for _,v:=range attachData { //子元素
+					if attach,isOK := v.(map[string]interface{});isOK {
+						if !dealWithRepeatAttachData(attach_text,attach) {
+							//符合条件-不重复直接添加
+							keyIndex++
+							saveKey := fmt.Sprintf("%v",keyIndex)
+							attach_text[saveKey] = attach //key累加
+							isAttach = true
+
+							//多条情况-融合
+							if (*recordDict)["attach_text"]==nil {
+								(*recordDict)["attach_text"] = []map[string]interface{}{
+									map[string]interface{}{
+										"id":value_id,
+										"value":attach,
+									},
+								}
+							}else {
+								arr := (*recordDict)["attach_text"].([]map[string]interface{})
+								arr = append(arr,map[string]interface{}{
+									"id":value_id,
+									"value":attach,
+								})
+								(*recordDict)["attach_text"] = arr
+							}
+
+						}
+					}
+				}
+			}
+		}
+	}
+
+
+
+	//联系人 winnerorder
+	winnerCount:=qu.IntAll(0)
+	winnerArr,b,isWinner,winnerid:=make(primitive.A,0),false,false,templateid
+	if winnerArr,b = templateTmp["winnerorder"].([]interface{});b {
+		winnerCount = qu.IntAll(len(winnerArr))
+	}
+
+
+	//分包 package
+	packageCount:=qu.IntAll(0)
+	packageArr,b,isPackage,packageid:=make(map[string]interface{},0),false,false,templateid
+	if packageArr,b = templateTmp["package"].(map[string]interface{});b {
+		packageCount = qu.IntAll(len(packageArr))
+	}
+
+	//遍历其他数据-
+	for _,value:=range weight.saveids {
+		if templateid == value {
+			continue
+		}
+		//winnerorder
+		tmp:=weight.data[value].data
+
+		if arr_1,winner_b := tmp["winnerorder"].(primitive.A);winner_b {
+			count:=qu.IntAll(len(arr_1))
+			if count > winnerCount {
+				winnerCount = count
+				winnerArr = arr_1
+				isWinner = true
+				winnerid = value
+			}
+		}
+		//package
+		if arr_2,package_b := (tmp["package"]).(map[string]interface{});package_b {
+			count:=qu.IntAll(len(arr_2))
+			if count > packageCount {
+				packageCount = count
+				packageArr = arr_2
+				isPackage = true
+				packageid = value
+			}
+		}
+	}
+
+
+	//改变的值
+	if len(winnerArr)>0 && winnerArr!=nil && isWinner {
+		modifyData["winnerorder"] = winnerArr
+		(*recordDict)["winnerorder"] = map[string]interface{}{
+			"id":winnerid,
+			"value":winnerArr,
+		}
+	}
+	if len(packageArr)>0 && packageArr!=nil && isPackage {
+		modifyData["package"] = packageArr
+		(*recordDict)["package"] = map[string]interface{}{
+			"id":packageid,
+			"value":packageArr,
+		}
+	}
+	if len(attach_text)>0 && attach_text!=nil && isAttach {
+		modifyData["attach_text"] = attach_text
+	}
+
+	return modifyData
+}
+

udpfusion/src/weightFusion.go

@@ -31,7 +31,7 @@ func (weight *weightDataMap) dealWithAddFusionStruct ()(map[string]interface{},m
 	//标准id
 	dict["fusion_templateid"] = weight.templateid
 	//站点，质量分
-	dict["fusion_score"] = weight.dealWithLevelAndScoreRecord()
+	dict["fusion_score"] = weight.saveLevelAndScoreRecord()
 	//记录所有href
 	dict["fusion_all_hrefs"] = weight.allhrefs
 	//采用新增id
@@ -64,7 +64,7 @@ func (weight *weightDataMap) dealWithMultipleAddFusionStruct ()(map[string]inter
 	//融合模板
 	dict["fusion_templateid"] = weight.templateid
 	//站点，质量分
-	dict["fusion_score"] = weight.dealWithLevelAndScoreRecord()
+	dict["fusion_score"] = weight.saveLevelAndScoreRecord()
 	//当前更新时间
 	dict["fusion_time"] = qu.IntAll(time.Now().Unix())
 	//融合生成时间
@@ -76,28 +76,24 @@ func (weight *weightDataMap) dealWithMultipleAddFusionStruct ()(map[string]inter
 
 	//采用新增id
 	delete(dict,"_id")
-	//日志记录
-	recordDict := make(map[string]interface{},0)
-	//结构体字段逻辑处理
-	structData := weight.dealWithStructData(&recordDict)
-	for k,v:=range structData {
-		dict[k] = v
-	}
-	//非空新增字段
-	otherFieldData := weight.dealWithOtherFieldData(&recordDict)
-	for k,v:=range otherFieldData {
-		dict[k] = v
+
+	//发生融合的数据
+	new_data := weight.dealWithAllFusionFieldData()
+	for k,v:=range new_data {
+		value:= *qu.ObjToMap(v)
+		dict[k] = value["value"]
 	}
+
+	//日志记录
 	newRecordDict := make(map[string]interface{},0)
 	newRecordDict["1"] = map[string]interface{}{
-		"data":recordDict,
+		"data":new_data,
 		"snapshot":map[string]interface{}{},
 	}
 	newRecordDict["number"] = qu.Int64All(1)
 
 	//融合表记录的字段
-	dict["fusion_fields"] = weight.dealWithFusionFields(recordDict)
-
+	dict["fusion_fields"] = weight.saveChangeFields(new_data)
 
 	//返回，更新数据，日志记录数据
 	return dict,newRecordDict
@@ -121,7 +117,7 @@ func (weight *weightDataMap) dealWithMultipleUpdateFusionStruct (tmpData map[str
 	//记录所有href
 	dict["fusion_all_hrefs"] = weight.allhrefs
 	//站点，质量分
-	dict["fusion_score"] = weight.dealWithLevelAndScoreRecord()
+	dict["fusion_score"] = weight.saveLevelAndScoreRecord()
 	//融合生成时间-取融合表第一次融合时间
 	dict["fusion_time"] = tmpData["fusion_time"]
 	//当前更新时间
@@ -129,63 +125,82 @@ func (weight *weightDataMap) dealWithMultipleUpdateFusionStruct (tmpData map[str
 	//删除_id
 	delete(dict,"_id")
 
-
-	//日志记录-还有快照页面 等等
-	recordDict := make(map[string]interface{},0)
-	//结构体字段逻辑处理
-	structData := weight.dealWithStructData(&recordDict)
-	for k,v:=range structData {
-		dict[k] = v
-	}
-	//非空新增字段
-	otherFieldData := weight.dealWithOtherFieldData(&recordDict)
-	for k,v:=range otherFieldData {
-		dict[k] = v
+	//发生融合的数据
+	new_data := weight.dealWithAllFusionFieldData()
+	for k,v:=range new_data {
+		value:= *qu.ObjToMap(v)
+		dict[k] = value["value"]
 	}
+
+	//日志记录-更新
 	newRecordDict := make(map[string]interface{},0)
 	newRecordDict = mgo.FindById(record_coll_name,BsonTOStringId(tmpData["_id"]))
 	number:=qu.Int64All(newRecordDict["number"])
 	number++
 	key:=fmt.Sprintf("%d",number)
 	newRecordDict[key] = map[string]interface{}{
-		"data":recordDict,
-		"snapshot":tmpData,
+		"data":new_data,
+		"snapshot":tmpData, //快照页面... ...
 	}
 	newRecordDict["number"] = number
 
 	//融合表记录的字段
-	dict["fusion_fields"] = weight.dealWithFusionFields(recordDict)
+	dict["fusion_fields"] = weight.saveChangeFields(new_data)
 
 	//返回，更新数据，日志记录数据
 	return dict,newRecordDict
 }
 
 
-/*
-	字段处理方法
-	**********************
-	**********************
-*/
-
-//处理其他字段数据
-func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interface{}) map[string]interface{} {
-
+/*字段处理方法*/
+func (weight *weightDataMap)dealWithAllFusionFieldData()map[string]interface{} {
 	//模板id 数据
 	templateid := weight.templateid
 	templateTmp := weight.data[templateid].data
 	modifyData := make(map[string]interface{}, 0) //返回修改的数据
-	//找到非空数据
-	arr := make([]string,0)
+	//前置处理 - 结构化数据 -  Arr
+	structArrData := *qu.ObjToMap(fusionAllKey["Arr"])
+	for key,_ := range structArrData {
+		count:=qu.IntAll(0)
+		arr,b,isOK,arr_id:=make(primitive.A,0),false,false,templateid
+		if arr,b = templateTmp[key].([]interface{});b {
+			count = qu.IntAll(len(arr))
+		}
+		for _,cur_id:=range weight.saveids {
+			if templateid == cur_id {
+				continue
+			}
+			tmp:=weight.data[cur_id].data
+			if arr_1,isTrue := tmp[key].(primitive.A);isTrue {
+				count_1:=qu.IntAll(len(arr_1))
+				if count_1 > count {
+					count = count_1
+					arr = arr_1
+					isOK = true
+					arr_id = cur_id
+				}
+			}
+		}
+		if len(arr)>0 && arr!=nil && isOK { //有改变的值
+			modifyData[key] = map[string]interface{}{
+				"id":arr_id,
+				"value":arr,
+			}
+		}
+	}
+
+
+	//第一步融合模板无效值
+	invalidKeyArr := make([]string,0) //无效
 	for key,value:=range templateTmp {
-		if judgeIsFusionKey(key) { //存在key且无效
-			if !judgeIsEffectiveData(value,key) {
-				arr = append(arr,key)
+		if judgeIsFusionKey(key) {
+			if !judgeIsEffectiveData(value,key) { //存在key且无效
+				invalidKeyArr = append(invalidKeyArr,key)
 			}
 		}
 	}
-	//第一步，替换模板，存在且空值
-	if arr!=nil && len(arr)>0 {
-		for _,key:=range arr  {
+	if invalidKeyArr!=nil && len(invalidKeyArr)>0 {
+		for _,key:=range invalidKeyArr  {
 			isRank := 2
 		L:	for {
 				for _,v:=range weight.saveids {
@@ -196,9 +211,8 @@ func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interf
 					if dataInfo.ranking==isRank { //找到指定排名-字段数据
 						value:=dataInfo.data[key]
 						if value !=nil && judgeIsEffectiveData(value,key)  {
-							modifyData[key] = value
 							templateTmp[key] = value
-							(*recordDict)[key] = map[string]interface{}{
+							modifyData[key] = map[string]interface{}{
 								"id":v,
 								"value":value,
 							}
@@ -206,7 +220,6 @@ func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interf
 						}
 						break
 					}
-
 				}
 				isRank++
 				if isRank > len(weight.saveids) {
@@ -215,7 +228,6 @@ func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interf
 			}
 		}
 	}
-	//log.Println("待替换key:",arr,"修改后:",modifyData)
 
 	//第二步-集合最大化
 	isRank := 2
@@ -224,21 +236,16 @@ func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interf
 			if v == templateid {
 				continue
 			}
-
-
 			dataInfo:=weight.data[v]
 			if dataInfo.ranking==isRank { //找到指定排名数据
 				for key,newValue:=range dataInfo.data{
-					if key=="_id" || templateTmp[key]!=nil || NoNeedFusionKey[key]!=nil{
-						continue
-					}
-					if judgeIsEffectiveData(newValue,key) {
-						//log.Println("最大化有效-",key)
-						templateTmp[key] = newValue
-						modifyData[key] = newValue
-						(*recordDict)[key] = map[string]interface{}{
-							"id":v,
-							"value":newValue,
+					if key!="_id" && judgeIsFusionKey(key) && templateTmp[key]==nil {
+						if judgeIsEffectiveData(newValue,key) {
+							templateTmp[key] = newValue
+							modifyData[key] = map[string]interface{}{
+								"id":v,
+								"value":newValue,
+							}
 						}
 					}
 				}
@@ -251,139 +258,6 @@ func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interf
 		}
 	}
 
-	//log.Println("isRank:",isRank,len(modifyData))
-
-	return modifyData
-}
-
-//处理结构数据
-func (weight *weightDataMap)dealWithStructData(recordDict *map[string]interface{}) map[string]interface{} {
-
-	//模板id 数据
-	templateid:=weight.templateid
-	templateTmp:=weight.data[templateid].data
-
-	modifyData :=make(map[string]interface{},0)
-
-	attach_text,isAttach:=make(map[string]interface{},0),false
-	if tmp_arr,b := templateTmp["attach_text"].(map[string]interface{});b {
-		//有值符合-
-		attach_text = tmp_arr
-	}
-	//附件判重-并合并新增
-	keyIndex := -1
-	for k,_:=range attach_text {
-		key:=qu.IntAll(k)
-		if key>keyIndex {
-			keyIndex = key
-		}
-	}
-	for _,value_id :=range weight.saveids {
-		if templateid == value_id {
-			continue
-		}
-		rankData := weight.data[value_id].data //具体其他排名数据
-		if attachData,b := rankData["attach_text"].(map[string]interface{});b {
-			if len(attachData)>0  { //有值
-				for _,v:=range attachData { //子元素
-					if attach,isOK := v.(map[string]interface{});isOK {
-						if !dealWithRepeatAttachData(attach_text,attach) {
-							//符合条件-不重复直接添加
-							keyIndex++
-							saveKey := fmt.Sprintf("%v",keyIndex)
-							attach_text[saveKey] = attach //key累加
-							isAttach = true
-
-							//多条情况-融合
-							if (*recordDict)["attach_text"]==nil {
-								(*recordDict)["attach_text"] = []map[string]interface{}{
-									map[string]interface{}{
-										"id":value_id,
-										"value":attach,
-									},
-								}
-							}else {
-								arr := (*recordDict)["attach_text"].([]map[string]interface{})
-								arr = append(arr,map[string]interface{}{
-									"id":value_id,
-									"value":attach,
-								})
-								(*recordDict)["attach_text"] = arr
-							}
-
-						}
-					}
-				}
-			}
-		}
-	}
-
-
-
-	//联系人 winnerorder
-	winnerCount:=qu.IntAll(0)
-	winnerArr,b,isWinner,winnerid:=make(primitive.A,0),false,false,templateid
-	if winnerArr,b = templateTmp["winnerorder"].([]interface{});b {
-		winnerCount = qu.IntAll(len(winnerArr))
-	}
-
-
-	//分包 package
-	packageCount:=qu.IntAll(0)
-	packageArr,b,isPackage,packageid:=make(map[string]interface{},0),false,false,templateid
-	if packageArr,b = templateTmp["package"].(map[string]interface{});b {
-		packageCount = qu.IntAll(len(packageArr))
-	}
-
-	//遍历其他数据-
-	for _,value:=range weight.saveids {
-		if templateid == value {
-			continue
-		}
-		//winnerorder
-		tmp:=weight.data[value].data
-
-		if arr_1,winner_b := tmp["winnerorder"].(primitive.A);winner_b {
-			count:=qu.IntAll(len(arr_1))
-			if count > winnerCount {
-				winnerCount = count
-				winnerArr = arr_1
-				isWinner = true
-				winnerid = value
-			}
-		}
-		//package
-		if arr_2,package_b := (tmp["package"]).(map[string]interface{});package_b {
-			count:=qu.IntAll(len(arr_2))
-			if count > packageCount {
-				packageCount = count
-				packageArr = arr_2
-				isPackage = true
-				packageid = value
-			}
-		}
-	}
-
-
-	//改变的值
-	if len(winnerArr)>0 && winnerArr!=nil && isWinner {
-		modifyData["winnerorder"] = winnerArr
-		(*recordDict)["winnerorder"] = map[string]interface{}{
-			"id":winnerid,
-			"value":winnerArr,
-		}
-	}
-	if len(packageArr)>0 && packageArr!=nil && isPackage {
-		modifyData["package"] = packageArr
-		(*recordDict)["package"] = map[string]interface{}{
-			"id":packageid,
-			"value":packageArr,
-		}
-	}
-	if len(attach_text)>0 && attach_text!=nil && isAttach {
-		modifyData["attach_text"] = attach_text
-	}
-
 	return modifyData
 }
 
@@ -394,7 +268,7 @@ func (weight *weightDataMap)dealWithStructData(recordDict *map[string]interface{
 	******************************
 	******************************
 */
-func (weight *weightDataMap) dealWithLevelAndScoreRecord () map[string]interface{}{
+func (weight *weightDataMap) saveLevelAndScoreRecord () map[string]interface{}{
 	dict := make(map[string]interface{},0)
 	saveids:= weight.saveids
 	for _,v:=range saveids{
@@ -406,7 +280,7 @@ func (weight *weightDataMap) dealWithLevelAndScoreRecord () map[string]interface
 	return dict
 }
 //处理-融合的哪些字段记录
-func (weight *weightDataMap)dealWithFusionFields(data map[string]interface{}) map[string]interface{} {
+func (weight *weightDataMap)saveChangeFields(data map[string]interface{}) map[string]interface{} {
 	fieldCal := make(map[string]interface{},0)
 	for k,v:=range data{
 		if k=="attach_text" {

udpfusion/src/weightFusionMethod.go

@@ -115,21 +115,20 @@ func sortTimeArrMethod(arr []int) (int,int) {
 	return arr[indexEarly],arr[indexLately]
 }
 
-//返回true 需要融合的数据
+//返回true 需要融合的字段
 func judgeIsFusionKey(key string) bool {
-	b:=false
-	if NoNeedFusionKey[key]==nil {
-		b=true
+	s_dict := *qu.ObjToMap(fusionAllKey["String"])
+	i_dict := *qu.ObjToMap(fusionAllKey["Int"])
+	f_dict := *qu.ObjToMap(fusionAllKey["Float"])
+	if s_dict[key]!=nil || i_dict[key]!=nil || f_dict[key]!=nil {
+		return true
 	}
-	return b
+	return false
 }
 
 func judgeIsEffectiveData(value interface{},key string) bool  {
-	if value==nil { //不存在的值
-		return false
-	}
 
-	//指定字段处理 area projectaddr buyeraddr agencyaddr winneraddr
+	//指定字段处理
 	if key=="area" {
 		if qu.ObjToString(value) == "全国" || qu.ObjToString(value) == ""  {
 			return false
@@ -137,34 +136,32 @@ func judgeIsEffectiveData(value interface{},key string) bool  {
 			return true
 		}
 	}
-	if key=="projectaddr"||key=="buyeraddr"||key=="agencyaddr"||key=="winneraddr" {
+	if key=="buyeraddr"||key=="agencyaddr"||key=="winneraddr" {
 		if qu.ObjToString(value) == "无" || qu.ObjToString(value) == ""  {
 			return false
 		}else {
 			return true
 		}
 	}
-
-	valueType := reflect.ValueOf(value).Kind()
 	//int float 类型
+	valueType := reflect.ValueOf(value).Kind()
 	if valueType==reflect.Int || valueType==reflect.Int32 || valueType==reflect.Int64 ||
 		valueType==reflect.Int8 || valueType==reflect.Int16||valueType==reflect.Float32||
 		valueType==reflect.Float64{
-		//如果类型为金额类
-		if key=="bidopentime" || key=="bidamount" || key=="budget" ||
-			key=="publishtime" || key=="comeintime" {
+		if key=="bidopentime" || key=="bidamount" || key=="budget" {
 			if qu.Float64All(value) == 0.0 {
 				return false
 			}
 		}
 		return true
 	}
+
 	//bool类型处理
 	if valueType==reflect.Bool {
 		return true
 	}
 
-	//其他类型采用
+	/* Array Chan Map Slice String*/
 	valueLen := reflect.ValueOf(value).Len()
 	if valueLen>0 {
 		return true

udpfusion/src/weightValue.go

@@ -1,8 +1,8 @@
 package main
 
 import (
-	qu "qfw/util"
 	"fmt"
+	qu "qfw/util"
 )
 type weightInfo struct {
 	maxLevel 		bool
@@ -30,11 +30,13 @@ func NewWeightData(arr []string) *weightDataMap {
 	for _,v:=range arr {
 		dict := mgo.FindById(coll_name,v)
 		if dict!=nil && len(dict)>2{
-			data[v] = analyzeTheSoureData(dict)
-			hrefs:= weight.allhrefs
+
 			keyid := fmt.Sprintf(BsonTOStringId(dict["_id"]))
+			hrefs:= weight.allhrefs
 			hrefs[keyid] = qu.ObjToString(dict["href"])
 			weight.allhrefs = hrefs
+
+			data[v] = analyzeTheSoureData(dict)
 		}
 	}