更新企业变更

qyxy_change_new/task.go

@@ -20,19 +20,20 @@ var mutex sync.Mutex
 //IncData 增量处理数据
 func IncData() {
 	defer util.Catch()
-	sess := Mgo.GetMgoConn()
+	sess := Mgo.GetMgoConn() //181 凭安
 	defer Mgo.DestoryMongoConn(sess)
 
-	q := bson.M{"jy_updatetime": bson.M{"$gt": jyUpdatetime}}
+	q := bson.M{"jy_updatetime": bson.M{"$gte": jyUpdatetime}}
 	var zid int
+	queryCount, _ := sess.DB("mixdata").C("company_change").Find(q).Count()
+	util.Debug("queryCount", queryCount)
 	it := sess.DB("mixdata").C("company_change").Find(q).Select(nil).Iter()
 	count := 0
-	ch := make(chan bool, 16)
-	wg := &sync.WaitGroup{}
-
+	realCount := 0
 	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
 		if count%20000 == 0 {
 			log.Println("current:", count)
+			log.Println("current: realCount", realCount)
 		}
 		if util.ObjToString(tmp["_operation_type"]) == "update" {
 			continue
@@ -43,62 +44,53 @@ func IncData() {
 		}
 
 		zid = util.IntAll(tmp["_id"])
-		ch <- true
-		wg.Add(1)
-		go func(tmp map[string]interface{}) {
-			defer func() {
-				<-ch
-				wg.Done()
-			}()
-			mutex.Lock()
-			defer mutex.Unlock()
-			//
-			currentTime := time.Now().Unix()
-			query := bson.M{"company_id": tmp["company_id"]}
-			info, b := MgoMix.FindOneByField(CollSave, query, bson.M{"changes": 1})
-			//原来数据有changes 字段就更新，追加数据
-			if b && len(*info) > 0 {
-				if util.ObjToString(tmp["_operation_type"]) == "insert" {
-					update := make(map[string]interface{})
-					item := make(map[string]interface{})
-					item["change_field"] = tmp["change_field"]
-					item["content_before"] = tmp["content_before"]
-					item["content_after"] = tmp["content_after"]
-					item["change_date"] = tmp["change_date"]
-					setMark(item) //change_name_new
-					//update["changes"] = changes
-					update["update_time"] = currentTime
-					saveInfo := map[string]interface{}{"$set": update, "$push": map[string]interface{}{"changes": item}}
-					MgoMix.Update("qyxy_change", map[string]interface{}{"company_id": util.ObjToString(tmp["company_id"])}, saveInfo, true, false)
-				}
-			} else {
-				//没有的直接写入
-				query := bson.M{"_id": tmp["company_id"]}
-				qyxy, b1 := MgoMix.FindOne("qyxy_std", query)
-				if b1 && len(*qyxy) > 0 {
-					save := make(map[string]interface{})
-					var changes []map[string]interface{}
-					item := make(map[string]interface{})
-					item["change_field"] = tmp["change_field"]
-					item["content_before"] = tmp["content_before"]
-					item["content_after"] = tmp["content_after"]
-					item["change_date"] = tmp["change_date"]
-					setMark(item) //change_name_new
-					changes = append(changes, item)
+		currentTime := time.Now().Unix()
+		query := bson.M{"company_id": tmp["company_id"]}
+		info, b := MgoMix.FindOneByField(CollSave, query, bson.M{"changes": 1})
+		//原来数据有changes 字段就更新，追加数据
+		if b && len(*info) > 0 {
+			realCount++
+			if util.ObjToString(tmp["_operation_type"]) == "insert" {
+				update := make(map[string]interface{})
+				item := make(map[string]interface{})
+				item["change_field"] = tmp["change_field"]
+				item["content_before"] = tmp["content_before"]
+				item["content_after"] = tmp["content_after"]
+				item["change_date"] = tmp["change_date"]
+				setMark(item) //change_name_new
+				//update["changes"] = changes
+				update["update_time"] = currentTime
+				saveInfo := map[string]interface{}{"$set": update, "$push": map[string]interface{}{"changes": item}}
+				MgoMix.Update(CollSave, map[string]interface{}{"company_id": util.ObjToString(tmp["company_id"])}, saveInfo, true, false)
+			}
+		} else {
+			//没有的直接写入
+			query := bson.M{"_id": tmp["company_id"]}
+			qyxy, b1 := MgoMix.FindOne("qyxy_std", query)
+			if b1 && len(*qyxy) > 0 {
+				realCount++
+				save := make(map[string]interface{})
+				var changes []map[string]interface{}
+				item := make(map[string]interface{})
+				item["change_field"] = tmp["change_field"]
+				item["content_before"] = tmp["content_before"]
+				item["content_after"] = tmp["content_after"]
+				item["change_date"] = tmp["change_date"]
+				setMark(item) //change_name_new
+				changes = append(changes, item)
 
-					//save["company_name"] = (*qyxy)["company_name"]
-					save["company_id"] = (*qyxy)["_id"]
-					save["changes"] = changes
-					save["create_time"] = currentTime
-					save["update_time"] = currentTime
-					//saveInfo := map[string]interface{}{"$set": save}
-					MgoMix.Save(CollSave, save)
-				}
+				//save["company_name"] = (*qyxy)["company_name"]
+				save["company_id"] = (*qyxy)["_id"]
+				save["changes"] = changes
+				save["create_time"] = currentTime
+				save["update_time"] = currentTime
+				//saveInfo := map[string]interface{}{"$set": save}
+				MgoMix.Save(CollSave, save)
 			}
-		}(tmp)
-		tmp = map[string]interface{}{}
+		}
+
 	}
-	wg.Wait()
+
 	util.Debug("over---", count, zid)
 
 }

udp/main.go

@@ -15,7 +15,7 @@ import (
 var startDate, endDate string
 
 func main() {
-	ip, start_time, end_time, p, tmptime, tmpkey, id1, id2, stype, q, bkey, param, ids, path, start, infoid := "", 0, 0, 0, 0, "", "", "", "", "", "", "", "", "", "", ""
+	ip, start_time, end_time, jy_updatetime, p, tmptime, tmpkey, id1, id2, stype, q, bkey, param, ids, path, start, infoid := "", 0, 0, 0, 0, 0, "", "", "", "", "", "", "", "", "", "", ""
 	flag.StringVar(&startDate, "startDate", "", "开始日期2006-01-02")
 	flag.StringVar(&endDate, "endDate", "", "结束日期2006-01-02")
 	flag.StringVar(&ip, "ip", "127.0.0.1", "ip")
@@ -23,6 +23,7 @@ func main() {
 	flag.IntVar(&tmptime, "tmptime", 0, "时间查询")
 	flag.IntVar(&start_time, "start_time", 0, "开始时间戳")
 	flag.IntVar(&end_time, "end_time", 0, "结束时间戳")
+	flag.IntVar(&jy_updatetime, "jy_updatetime", 0, "qyxy_change更新数据时间")
 	flag.StringVar(&tmpkey, "tmpkey", "", "时间字段")
 	flag.StringVar(&id1, "gtid", "", "gtid")
 	flag.StringVar(&id2, "lteid", "", "lteid")
@@ -85,6 +86,10 @@ func main() {
 			m1["start_time"] = start_time
 		}
 
+		if jy_updatetime > 0 {
+			m1["jy_updatetime"] = jy_updatetime
+		}
+
 		if end_time > 0 {
 			m1["end_time"] = end_time
 		}