更新dataging

udpfilterdup/src/README.md

@@ -1,8 +1,8 @@
 基于内存的信息重复过滤
 
 {
-    "udpport": ":1485",
-    "dupdays": 5,
+    "udpport": ":11485",
+    "dupdays": 7,
     "mongodb": {
         "addr": "172.17.4.187:27083",
         "pool": 10,
@@ -32,13 +32,14 @@
             "memo": "创建招标数据索引new"
         }
     ],
+
     "threads": 1,
     "isMerger": false,
     "isSort":false,
     "lowHeavy":false,
     "timingTask":true,
-    "timingSpanDay": 2,
-    "timingPubScope": 360,
+    "timingSpanDay": 3,
+    "timingPubScope": 720,
     "specialwords": "(重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研)",
     "specialtitle_0": "(包|标段|标包)[(（]?[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]?",
     "specialtitle_1": "[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９](次|包|标段|标包|批)",
@@ -51,3 +52,10 @@
 
 
 
+
+
+
+
+
+
+

udpfilterdup/src/config.json

@@ -36,17 +36,11 @@
     "isSort":false,
     "lowHeavy":false,
     "timingTask":true,
-    "timingSpanDay": 2,
-    "timingPubScope": 720,
+    "timingSpanDay": 3,
+    "timingPubScope": 1080,
     "specialwords": "(重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研)",
     "specialtitle_0": "(包|标段|标包)[(（]?[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]?",
     "specialtitle_1": "[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９](次|包|标段|标包|批)",
     "specialtitle_2": "项目[(（][0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]",
     "beifen": "[(（]?[0-9一二三四五六七八九十零１２３４５６７８９再][)）]?[子分]?[次批标包]|重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研"
-}
-
-
-
-
-
-
+}

udpfilterdup/src/main.go

@@ -107,9 +107,6 @@ func init() {
 
 func main() {
 
-	//TestXiuFu()
-	//
-	//return
 	go checkMapJob()
 	updport := Sysconfig["udpport"].(string)
 	udpclient = mu.UdpClient{Local: updport, BufSize: 1024}
@@ -691,6 +688,7 @@ func timedTaskOnce() {
 	defer mgo.DestoryMongoConn(sess)
 	it_start := sess.DB(mgo.DbName).C(extract).Find(&q_start).Sort("publishtime").Iter()
 	num,oknum, deterTime:= 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_start.Next(&tmp); num++ {
@@ -717,11 +715,35 @@ func timedTaskOnce() {
 						dayArr = append(dayArr,tmp)
 					}
 				}
+			}else {
+				//不在两年内的也清标记
+				updateExtract = append(updateExtract, []map[string]interface{}{
+					map[string]interface{}{
+						"_id": tmp["_id"],
+					},
+					map[string]interface{}{
+						"$set": map[string]interface{}{
+							"dataging": 0,
+						},
+					},
+				})
+				if len(updateExtract) > 50 {
+					mgo.UpSertBulk(extract, updateExtract...)
+					updateExtract = [][]map[string]interface{}{}
+				}
+
 			}
 		}
 		tmp = make(map[string]interface{})
 	}
 
+
+	//批量更新标记
+	if len(updateExtract) > 0 {
+		mgo.UpSertBulk(extract, updateExtract...)
+		updateExtract = [][]map[string]interface{}{}
+	}
+
 	if len(dayArr)>0 {
 		pendAllArr = append(pendAllArr,dayArr)
 		dayArr = []map[string]interface{}{}
@@ -742,8 +764,6 @@ func timedTaskOnce() {
 	}
 	log.Println("本地构建分组完成:",len(pendAllArr),"组","测试-总计数量：",testNum)
 
-
-	updateExtract := [][]map[string]interface{}{}
 	n, repeateN := 0, 0
 	for k,v:=range pendAllArr {
 		//构建当前组的数据池
@@ -768,7 +788,7 @@ func timedTaskOnce() {
 							},
 						},
 					})
-					if len(updateExtract) > 10 {
+					if len(updateExtract) > 50 {
 						mgo.UpSertBulk(extract, updateExtract...)
 						updateExtract = [][]map[string]interface{}{}
 					}
@@ -805,7 +825,7 @@ func timedTaskOnce() {
 					},
 				})
 			}
-			if len(updateExtract) > 10 {
+			if len(updateExtract) > 50 {
 				mgo.UpSertBulk(extract, updateExtract...)
 				updateExtract = [][]map[string]interface{}{}
 			}
@@ -816,6 +836,7 @@ func timedTaskOnce() {
 
 	if len(updateExtract) > 0 {
 		mgo.UpSertBulk(extract, updateExtract...)
+		updateExtract = [][]map[string]interface{}{}
 	}
 	log.Println("this timeTask over.", n, "repeateN:", repeateN)