{
    "udpport": ":19097",
    "dupdays": 7,
    "mongodb": {
        "addr": "172.17.4.85:27080",
        "pool": 10,
        "db": "qfw",
        "extract": "result_20200715",
        "extract_back": "result_20200714",
        "site": {
            "dbname": "qfw",
            "coll": "site"
        }
    },
    "task_mongodb": {
        "task_addrName": "172.17.4.187:27081",
        "task_dbName": "qfw",
        "task_collName": "ocr_flie_over",
        "pool": 5
        
    },
    "jkmail": {
        "to": "zhengkun@topnet.net.cn,zhangjinkun@topnet.net.cn",
        "api": "http://10.171.112.160:19281/_send/_mail"
    },
    "nextNode": [
        {
            "addr": "172.17.4.194",
            "port": 1782,
            "stype": "project",
            "memo": "合并项目"
        },
        {
            "addr": "127.0.0.1",
            "port": 1783,
            "stype": "bidding",
            "memo": "创建招标数据索引new"
        }
    ],
    "threads": 1,
    "isMerger": false,
    "lowHeavy":true,
    "timingTask":false,
    "timingSpanDay": 5,
    "timingPubScope": 720,
    "specialwords": "(重招|重新招标|勘察|施工|监理|总承包|土石方|可研)",
    "specialtitle_0": "(包|标段|标包)[(（]?[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]?",
    "specialtitle_1": "[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９](次|包|标段|标包|批|期)",
    "specialtitle_2": "项目[(（][0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]",
    "beifen": "[(（]?[0-9一二三四五六七八九十零１２３４５６７８９再][)）]?[子分]?[次批标包]|重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研"
}






mgo = &MongodbSim{
		MongodbAddr: "172.17.4.187:27083",
		DbName:      "qfw",
		Size:        10,
	}
mgo.InitPool()
	return
	
func moveTimeoutData()  {
	log.Println("部署迁移定时任务")
	c := cron.New()
	c.AddFunc("0 0 0 * * ?", func() { moveOnceTimeOut() })
	c.Start()
}

func moveOnceTimeOut()  {
	log.Println("执行一次迁移超时数据")

	sess := mgo.GetMgoConn()
	defer mgo.DestoryMongoConn(sess)
	now:=time.Now()
	move_time := time.Date(now.Year()-2, now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local).Unix()
	q := map[string]interface{}{
		"comeintime": map[string]interface{}{
			"$lt": move_time,
		},
	}
	log.Println(q)
	it := sess.DB(mgo.DbName).C("result_20200714").Find(&q).Iter()
	index := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
		if index%10000 == 0 {
			log.Println("index", index)
		}
		del_id:=BsonTOStringId(tmp["_id"])
		mgo.Save("result_20200713", tmp)
		mgo.DeleteById("result_20200714",del_id)
		tmp = map[string]interface{}{}
	}
	log.Println("save and delete", " ok index", index)
}





{
    "udpport": ":1785",
    "dupdays": 7,
    "mongodb": {
        "addr": "172.17.4.85:27080",
        "pool": 10,
        "db": "qfw",
        "extract": "result_20200715",
        "extract_back": "result_20200714",
        "site": {
            "dbname": "qfw",
            "coll": "site"
        }
    },
    "jkmail": {
        "to": "zhengkun@topnet.net.cn,zhangjinkun@topnet.net.cn",
        "api": "http://172.17.145.179:19281/_send/_mail"
    },
    "nextNode": [
        {
            "addr": "127.0.0.1",
            "port": 1783,
            "stype": "bidding",
            "memo": "创建招标数据索引new"
        }
    ],
    "threads": 1,
    "isMerger": false,
    "lowHeavy":true,
    "timingTask":false,
    "timingSpanDay": 5,
    "timingPubScope": 720,
    "specialwords": "(重招|重新招标|勘察|施工|监理|总承包|土石方|可研)",
    "specialtitle_0": "(包|标段|标包)[(（]?[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]?",
    "specialtitle_1": "[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９](次|包|标段|标包|批|期)",
    "specialtitle_2": "项目[(（][0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]",
    "beifen": "[(（]?[0-9一二三四五六七八九十零１２３４５６７８９再][)）]?[子分]?[次批标包]|重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研"
}

func historyTaskDay() {
	defer util.Catch()

	for {
		start:=time.Now().Unix()

		if gtid=="" {
			log.Println("请传gtid,否则无法运行")
			os.Exit(0)
			return
		}
		if lteid!="" {
			//先进行数据迁移
			log.Println("开启一次迁移任务",gtid,lteid)
			moveHistoryData(gtid,lteid)
			gtid = lteid //替换数据
		}

		//查询表最后一个id
		task_sess := task_mgo.GetMgoConn()
		defer task_mgo.DestoryMongoConn(task_sess)
		q:=map[string]interface{}{
			"isused":true,
		}
		between_time := time.Now().Unix() - (86400 * timingPubScope)//两年周期
		it_last := task_sess.DB(task_mgo.DbName).C(task_collName).Find(&q).Sort("-_id").Iter()
		for tmp := make(map[string]interface{}); it_last.Next(&tmp); {
			lteid = util.ObjToString(tmp["gtid"])
			log.Println("查询的最后一个任务Id:",lteid)
			break
		}

		log.Println("查询完毕-先睡眠5分钟",gtid,lteid)
		time.Sleep(5 * time.Minute)

		sess := mgo.GetMgoConn()//连接器
		defer mgo.DestoryMongoConn(sess)
		//开始判重
		q = map[string]interface{}{
			"_id": map[string]interface{}{
				"$gt": StringTOBsonId(gtid),
				"$lte": StringTOBsonId(lteid),
			},
		}
		log.Println("历史判重查询条件：",q,"时间：", between_time)
		it := sess.DB(mgo.DbName).C(extract).Find(&q).Sort("publishtime").Iter()
		num,oknum,outnum, deterTime:= int64(0),int64(0),int64(0),int64(0) //计数
		updateExtract := [][]map[string]interface{}{}//批量更新mongo数组
		pendAllArr:=[][]map[string]interface{}{}//待处理数组
		dayArr := []map[string]interface{}{}
		for tmp := make(map[string]interface{}); it.Next(&tmp); num++ {
			if num%10000 == 0 {
				log.Println("正序遍历:", num)
			}
			source := util.ObjToMap(tmp["jsondata"])
			if util.IntAll((*source)["sourcewebsite"]) == 1 {
				outnum++
				updatelock.Lock()
				updateExtract = append(updateExtract, []map[string]interface{}{
					map[string]interface{}{
						"_id": tmp["_id"],
					},
					map[string]interface{}{
						"$set": map[string]interface{}{
							"repeat": 1,
							"dataging": 0,
							"repeat_reason": "sourcewebsite为1 重复",
						},
					},
				})
				if len(updateExtract) >= 200 {
					log.Println("sourcewebsite,批量更新")
					mgo.UpSertBulk(extract, updateExtract...)
					updateExtract = [][]map[string]interface{}{}
				}

				updatelock.Unlock()


				tmp = make(map[string]interface{})
				continue
			}

			//取-符合-发布时间X年内的数据
			updatelock.Lock()
			if util.IntAll(tmp["dataging"]) == 1 {
				pubtime := util.Int64All(tmp["publishtime"])
				if pubtime > 0 && pubtime >= between_time {
					oknum++
					if deterTime==0 {
						log.Println("找到第一条符合条件的数据")
						deterTime = util.Int64All(tmp["publishtime"])
						dayArr = append(dayArr,tmp)
					}else {
						if pubtime-deterTime >timingSpanDay*86400 {
							//新数组重新构建，当前组数据加到全部组数据
							pendAllArr = append(pendAllArr,dayArr)
							dayArr = []map[string]interface{}{}
							deterTime = util.Int64All(tmp["publishtime"])
							dayArr = append(dayArr,tmp)
						}else {
							dayArr = append(dayArr,tmp)
						}
					}
				}else {
					outnum++
					//不在两年内的也清标记
					updateExtract = append(updateExtract, []map[string]interface{}{
						map[string]interface{}{
							"_id": tmp["_id"],
						},
						map[string]interface{}{
							"$set": map[string]interface{}{
								"dataging": 0,
							},
						},
					})
					if len(updateExtract) >= 200 {
						log.Println("不在周期内符合dataging==1,批量更新")
						mgo.UpSertBulk(extract, updateExtract...)
						updateExtract = [][]map[string]interface{}{}
					}

				}
			}

			updatelock.Unlock()

			tmp = make(map[string]interface{})
		}


		//批量更新标记
		updatelock.Lock()

		if len(updateExtract) > 0 {
			log.Println("分组后,最后更新不进行判重的数据：",len(updateExtract),oknum+outnum)
			mgo.UpSertBulk(extract, updateExtract...)
			updateExtract = [][]map[string]interface{}{}
		}

		updatelock.Unlock()


		if len(dayArr)>0 {
			pendAllArr = append(pendAllArr,dayArr)
			dayArr = []map[string]interface{}{}
		}

		log.Println("查询数量：",num,"符合条件：",oknum)

		if len(pendAllArr) <= 0 {
			log.Println("没找到dataging==1的数据")
		}

		//测试分组数量是否正确
		testNum:=0
		for k,v:=range pendAllArr {
			log.Println("第",k,"组--","数量：",len(v))
			testNum = testNum+len(v)
		}
		log.Println("本地构建分组完成:",len(pendAllArr),"组","测试-总计数量：",testNum)

		n, repeateN := 0, 0
		log.Println("线程数：",threadNum)
		pool := make(chan bool, threadNum)
		wg := &sync.WaitGroup{}
		for k,v:=range pendAllArr { //每组结束更新一波数据
			pool <- true
			wg.Add(1)
			go func(k int, v []map[string]interface{}) {
				defer func() {
					<-pool
					wg.Done()
				}()
				//每组临时数组 -  互不干扰
				groupUpdateExtract := [][]map[string]interface{}{}
				//
				groupOtherExtract := [][]map[string]interface{}{}

				//构建当前组的数据池
				log.Println("构建第", k, "组---（数据池）")
				//当前组的第一个发布时间
				first_pt := util.Int64All(v[len(v)-1]["publishtime"])
				curTM := TimedTaskDatamap(dupdays+int(timingSpanDay)+1, first_pt+86400, int(k))
				log.Println("开始遍历判重第", k, "组  共计数量：", len(v))
				n = n + len(v)
				log.Println("统计目前总数量：", n, "重复数量：", repeateN)
				for _, tmp := range v {
					info := NewInfo(tmp)
					b, source, reason := curTM.check(info)
					if b { //有重复,生成更新语句，更新抽取和更新招标
						repeateN++
						//重复数据打标签
						repeat_ids:=source.repeat_ids
						repeat_ids =  append(repeat_ids,info.id)
						source.repeat_ids = repeat_ids
						//替换数据池-更新
						DM.replacePoolData(source)
						updatelock.Lock()


						//更新数据源-   14 或者 15
						//判断是否在当前段落
						if judgeIsCurIds(gtid,lteid,source.id) {
							groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{//重复数据打标签
								map[string]interface{}{
									"_id": StringTOBsonId(source.id),
								},
								map[string]interface{}{
									"$set": map[string]interface{}{
										"repeat_ids": repeat_ids,
									},
								},
							})
						}else {
							groupOtherExtract = append(groupOtherExtract, []map[string]interface{}{//重复数据打标签
								map[string]interface{}{
									"_id": StringTOBsonId(source.id),
								},
								map[string]interface{}{
									"$set": map[string]interface{}{
										"repeat_ids": repeat_ids,
									},
								},
							})
						}
						groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
							map[string]interface{}{
								"_id": tmp["_id"],
							},
							map[string]interface{}{
								"$set": map[string]interface{}{
									"repeat":        1,
									"repeat_reason": reason,
									"repeat_id":     source.id,
									"dataging":      0,
								},
							},
						})

						if len(groupUpdateExtract) >= 500 {
							mgo.UpSertBulk(extract, groupUpdateExtract...)
							groupUpdateExtract = [][]map[string]interface{}{}
						}

						if len(groupOtherExtract) >= 500 {
							mgo.UpSertBulk(extract_back, groupOtherExtract...)
							groupOtherExtract = [][]map[string]interface{}{}
						}

						updatelock.Unlock()


					} else {
						updatelock.Lock()

						groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
							map[string]interface{}{
								"_id": tmp["_id"],
							},
							map[string]interface{}{
								"$set": map[string]interface{}{
									"dataging": 0, //符合条件的都为dataging==0
								},
							},
						})

						if len(groupUpdateExtract) >= 500 {
							mgo.UpSertBulk(extract, groupUpdateExtract...)
							groupUpdateExtract = [][]map[string]interface{}{}
						}
						updatelock.Unlock()
					}
				}
				//每组数据结束-更新数据
				updatelock.Lock()
				if len(groupUpdateExtract) > 0 {
					mgo.UpSertBulk(extract, groupUpdateExtract...)
				}

				if len(groupOtherExtract) > 0 {
					mgo.UpSertBulk(extract_back, groupOtherExtract...)
				}
				updatelock.Unlock()

			}(k, v)

		}

		wg.Wait()


		//任务完成，开始发送广播通知下面节点 发udp 去升索引待定 + 合并
		if n >= repeateN && gtid!=lteid{
			for _, to := range nextNode {
				next_sid := util.BsonIdToSId(gtid)
				next_eid := util.BsonIdToSId(lteid)
				key := next_sid + "-" + next_eid + "-" + util.ObjToString(to["stype"])
				by, _ := json.Marshal(map[string]interface{}{
					"gtid":  next_sid,
					"lteid": next_eid,
					"stype": util.ObjToString(to["stype"]),
					"key":   key,
				})
				addr := &net.UDPAddr{
					IP:   net.ParseIP(to["addr"].(string)),
					Port: util.IntAll(to["port"]),
				}
				node := &udpNode{by, addr, time.Now().Unix(), 0}
				udptaskmap.Store(key, node)
				udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
			}
		}

		end:=time.Now().Unix()

		log.Println("this timeTask over.", n, "repeateN:", repeateN,gtid,lteid)
		log.Println(gtid,lteid)
		if end-start<60*5 {
			log.Println("睡眠.............")
			time.Sleep(5 * time.Minute)
		}
		log.Println("继续下一段的历史判重")
	}
}func historyTaskDay() {
 	defer util.Catch()
 
 	for {
 		start:=time.Now().Unix()
 
 		if gtid=="" {
 			log.Println("请传gtid,否则无法运行")
 			os.Exit(0)
 			return
 		}
 		if lteid!="" {
 			//先进行数据迁移
 			log.Println("开启一次迁移任务",gtid,lteid)
 			moveHistoryData(gtid,lteid)
 			gtid = lteid //替换数据
 		}
 
 		//查询表最后一个id
 		task_sess := task_mgo.GetMgoConn()
 		defer task_mgo.DestoryMongoConn(task_sess)
 		q:=map[string]interface{}{
 			"isused":true,
 		}
 		between_time := time.Now().Unix() - (86400 * timingPubScope)//两年周期
 		it_last := task_sess.DB(task_mgo.DbName).C(task_collName).Find(&q).Sort("-_id").Iter()
 		for tmp := make(map[string]interface{}); it_last.Next(&tmp); {
 			lteid = util.ObjToString(tmp["gtid"])
 			log.Println("查询的最后一个任务Id:",lteid)
 			break
 		}
 
 		log.Println("查询完毕-先睡眠5分钟",gtid,lteid)
 		time.Sleep(5 * time.Minute)
 
 		sess := mgo.GetMgoConn()//连接器
 		defer mgo.DestoryMongoConn(sess)
 		//开始判重
 		q = map[string]interface{}{
 			"_id": map[string]interface{}{
 				"$gt": StringTOBsonId(gtid),
 				"$lte": StringTOBsonId(lteid),
 			},
 		}
 		log.Println("历史判重查询条件：",q,"时间：", between_time)
 		it := sess.DB(mgo.DbName).C(extract).Find(&q).Sort("publishtime").Iter()
 		num,oknum,outnum, deterTime:= int64(0),int64(0),int64(0),int64(0) //计数
 		updateExtract := [][]map[string]interface{}{}//批量更新mongo数组
 		pendAllArr:=[][]map[string]interface{}{}//待处理数组
 		dayArr := []map[string]interface{}{}
 		for tmp := make(map[string]interface{}); it.Next(&tmp); num++ {
 			if num%10000 == 0 {
 				log.Println("正序遍历:", num)
 			}
 			source := util.ObjToMap(tmp["jsondata"])
 			if util.IntAll((*source)["sourcewebsite"]) == 1 {
 				outnum++
 				updatelock.Lock()
 				updateExtract = append(updateExtract, []map[string]interface{}{
 					map[string]interface{}{
 						"_id": tmp["_id"],
 					},
 					map[string]interface{}{
 						"$set": map[string]interface{}{
 							"repeat": 1,
 							"dataging": 0,
 							"repeat_reason": "sourcewebsite为1 重复",
 						},
 					},
 				})
 				if len(updateExtract) >= 200 {
 					log.Println("sourcewebsite,批量更新")
 					mgo.UpSertBulk(extract, updateExtract...)
 					updateExtract = [][]map[string]interface{}{}
 				}
 
 				updatelock.Unlock()
 
 
 				tmp = make(map[string]interface{})
 				continue
 			}
 
 			//取-符合-发布时间X年内的数据
 			updatelock.Lock()
 			if util.IntAll(tmp["dataging"]) == 1 {
 				pubtime := util.Int64All(tmp["publishtime"])
 				if pubtime > 0 && pubtime >= between_time {
 					oknum++
 					if deterTime==0 {
 						log.Println("找到第一条符合条件的数据")
 						deterTime = util.Int64All(tmp["publishtime"])
 						dayArr = append(dayArr,tmp)
 					}else {
 						if pubtime-deterTime >timingSpanDay*86400 {
 							//新数组重新构建，当前组数据加到全部组数据
 							pendAllArr = append(pendAllArr,dayArr)
 							dayArr = []map[string]interface{}{}
 							deterTime = util.Int64All(tmp["publishtime"])
 							dayArr = append(dayArr,tmp)
 						}else {
 							dayArr = append(dayArr,tmp)
 						}
 					}
 				}else {
 					outnum++
 					//不在两年内的也清标记
 					updateExtract = append(updateExtract, []map[string]interface{}{
 						map[string]interface{}{
 							"_id": tmp["_id"],
 						},
 						map[string]interface{}{
 							"$set": map[string]interface{}{
 								"dataging": 0,
 							},
 						},
 					})
 					if len(updateExtract) >= 200 {
 						log.Println("不在周期内符合dataging==1,批量更新")
 						mgo.UpSertBulk(extract, updateExtract...)
 						updateExtract = [][]map[string]interface{}{}
 					}
 
 				}
 			}
 
 			updatelock.Unlock()
 
 			tmp = make(map[string]interface{})
 		}
 
 
 		//批量更新标记
 		updatelock.Lock()
 
 		if len(updateExtract) > 0 {
 			log.Println("分组后,最后更新不进行判重的数据：",len(updateExtract),oknum+outnum)
 			mgo.UpSertBulk(extract, updateExtract...)
 			updateExtract = [][]map[string]interface{}{}
 		}
 
 		updatelock.Unlock()
 
 
 		if len(dayArr)>0 {
 			pendAllArr = append(pendAllArr,dayArr)
 			dayArr = []map[string]interface{}{}
 		}
 
 		log.Println("查询数量：",num,"符合条件：",oknum)
 
 		if len(pendAllArr) <= 0 {
 			log.Println("没找到dataging==1的数据")
 		}
 
 		//测试分组数量是否正确
 		testNum:=0
 		for k,v:=range pendAllArr {
 			log.Println("第",k,"组--","数量：",len(v))
 			testNum = testNum+len(v)
 		}
 		log.Println("本地构建分组完成:",len(pendAllArr),"组","测试-总计数量：",testNum)
 
 		n, repeateN := 0, 0
 		log.Println("线程数：",threadNum)
 		pool := make(chan bool, threadNum)
 		wg := &sync.WaitGroup{}
 		for k,v:=range pendAllArr { //每组结束更新一波数据
 			pool <- true
 			wg.Add(1)
 			go func(k int, v []map[string]interface{}) {
 				defer func() {
 					<-pool
 					wg.Done()
 				}()
 				//每组临时数组 -  互不干扰
 				groupUpdateExtract := [][]map[string]interface{}{}
 				//
 				groupOtherExtract := [][]map[string]interface{}{}
 
 				//构建当前组的数据池
 				log.Println("构建第", k, "组---（数据池）")
 				//当前组的第一个发布时间
 				first_pt := util.Int64All(v[len(v)-1]["publishtime"])
 				curTM := TimedTaskDatamap(dupdays+int(timingSpanDay)+1, first_pt+86400, int(k))
 				log.Println("开始遍历判重第", k, "组  共计数量：", len(v))
 				n = n + len(v)
 				log.Println("统计目前总数量：", n, "重复数量：", repeateN)
 				for _, tmp := range v {
 					info := NewInfo(tmp)
 					b, source, reason := curTM.check(info)
 					if b { //有重复,生成更新语句，更新抽取和更新招标
 						repeateN++
 						//重复数据打标签
 						repeat_ids:=source.repeat_ids
 						repeat_ids =  append(repeat_ids,info.id)
 						source.repeat_ids = repeat_ids
 						//替换数据池-更新
 						DM.replacePoolData(source)
 						updatelock.Lock()
 
 
 						//更新数据源-   14 或者 15
 						//判断是否在当前段落
 						if judgeIsCurIds(gtid,lteid,source.id) {
 							groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{//重复数据打标签
 								map[string]interface{}{
 									"_id": StringTOBsonId(source.id),
 								},
 								map[string]interface{}{
 									"$set": map[string]interface{}{
 										"repeat_ids": repeat_ids,
 									},
 								},
 							})
 						}else {
 							groupOtherExtract = append(groupOtherExtract, []map[string]interface{}{//重复数据打标签
 								map[string]interface{}{
 									"_id": StringTOBsonId(source.id),
 								},
 								map[string]interface{}{
 									"$set": map[string]interface{}{
 										"repeat_ids": repeat_ids,
 									},
 								},
 							})
 						}
 						groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
 							map[string]interface{}{
 								"_id": tmp["_id"],
 							},
 							map[string]interface{}{
 								"$set": map[string]interface{}{
 									"repeat":        1,
 									"repeat_reason": reason,
 									"repeat_id":     source.id,
 									"dataging":      0,
 								},
 							},
 						})
 
 						if len(groupUpdateExtract) >= 500 {
 							mgo.UpSertBulk(extract, groupUpdateExtract...)
 							groupUpdateExtract = [][]map[string]interface{}{}
 						}
 
 						if len(groupOtherExtract) >= 500 {
 							mgo.UpSertBulk(extract_back, groupOtherExtract...)
 							groupOtherExtract = [][]map[string]interface{}{}
 						}
 
 						updatelock.Unlock()
 
 
 					} else {
 						updatelock.Lock()
 
 						groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
 							map[string]interface{}{
 								"_id": tmp["_id"],
 							},
 							map[string]interface{}{
 								"$set": map[string]interface{}{
 									"dataging": 0, //符合条件的都为dataging==0
 								},
 							},
 						})
 
 						if len(groupUpdateExtract) >= 500 {
 							mgo.UpSertBulk(extract, groupUpdateExtract...)
 							groupUpdateExtract = [][]map[string]interface{}{}
 						}
 						updatelock.Unlock()
 					}
 				}
 				//每组数据结束-更新数据
 				updatelock.Lock()
 				if len(groupUpdateExtract) > 0 {
 					mgo.UpSertBulk(extract, groupUpdateExtract...)
 				}
 
 				if len(groupOtherExtract) > 0 {
 					mgo.UpSertBulk(extract_back, groupOtherExtract...)
 				}
 				updatelock.Unlock()
 
 			}(k, v)
 
 		}
 
 		wg.Wait()
 
 
 		//任务完成，开始发送广播通知下面节点 发udp 去升索引待定 + 合并
 		if n >= repeateN && gtid!=lteid{
 			for _, to := range nextNode {
 				next_sid := util.BsonIdToSId(gtid)
 				next_eid := util.BsonIdToSId(lteid)
 				key := next_sid + "-" + next_eid + "-" + util.ObjToString(to["stype"])
 				by, _ := json.Marshal(map[string]interface{}{
 					"gtid":  next_sid,
 					"lteid": next_eid,
 					"stype": util.ObjToString(to["stype"]),
 					"key":   key,
 				})
 				addr := &net.UDPAddr{
 					IP:   net.ParseIP(to["addr"].(string)),
 					Port: util.IntAll(to["port"]),
 				}
 				node := &udpNode{by, addr, time.Now().Unix(), 0}
 				udptaskmap.Store(key, node)
 				udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
 			}
 		}
 
 		end:=time.Now().Unix()
 
 		log.Println("this timeTask over.", n, "repeateN:", repeateN,gtid,lteid)
 		log.Println(gtid,lteid)
 		if end-start<60*5 {
 			log.Println("睡眠.............")
 			time.Sleep(5 * time.Minute)
 		}
 		log.Println("继续下一段的历史判重")
 	}
 }