修改

standardata/src/historyrepair.go

@@ -0,0 +1,67 @@
+// historyrepair 处理多线程重复数据问题
+package main
+
+import (
+	"dbutil/mongo"
+	"dbutil/redis"
+	"log"
+	qu "qfw/util"
+
+	"go.mongodb.org/mongo-driver/bson"
+)
+
+func historyrepair(db, coll, datatype string, dbnum int) {
+	sess := MongoTo.GetMgoConn()
+	defer MongoTo.Close()
+	field := ""
+	if datatype == "winner" {
+		field = "company_name"
+	} else if datatype == "buyer" {
+		field = "buyer_name"
+	} else if datatype == "agency" {
+		field = "agency_name"
+	}
+	it := sess.DB(db).C(coll).Find(bson.M{}).Select(bson.M{field: 1}).Iter()
+	index := 0
+	delnum := 0
+	for tmp := map[string]interface{}{}; it.Next(&tmp); index++ {
+		name := qu.ObjToString(tmp[field])
+		id := mongo.BsonTOStringId(tmp["_id"])
+		str, _ := redis.GetRedisStr(datatype, dbnum, name)
+		if str != "" {
+			MongoTo.DeleteById(coll, id)
+			delnum++
+		} else {
+			redis.PutRedis(datatype, dbnum, name, id, -1)
+		}
+		tmp = map[string]interface{}{}
+		if index%100 == 0 {
+			log.Println(index, delnum)
+		}
+	}
+	log.Println(index, delnum)
+}
+
+func historyrepairErr(db, coll, datatype string, dbnum int) {
+	sess := MongoTo.GetMgoConn()
+	defer MongoTo.Close()
+	it := sess.DB(db).C(coll).Find(bson.M{}).Select(bson.M{"name": 1}).Iter()
+	index := 0
+	delnum := 0
+	for tmp := map[string]interface{}{}; it.Next(&tmp); index++ {
+		name := qu.ObjToString(tmp["name"])
+		id := mongo.BsonTOStringId(tmp["_id"])
+		str, _ := redis.GetRedisStr(datatype, dbnum, name)
+		if str != "" {
+			MongoTo.DeleteById(coll, id)
+			delnum++
+		} else {
+			redis.PutRedis(datatype, dbnum, name, id, -1)
+		}
+		tmp = map[string]interface{}{}
+		if index%100 == 0 {
+			log.Println(index, delnum)
+		}
+	}
+	log.Println(index, delnum)
+}

util/src/dbutil/mongo/mgo.go

@@ -146,7 +146,7 @@ func (m *MongodbSim) InitPool() {
 	opts := options.Client()
 	opts.SetConnectTimeout(3 * time.Second)
 	opts.ApplyURI("mongodb://" + m.MongodbAddr)
-	opts.SetMaxPoolSize(uint16(m.Size))
+	opts.SetMaxPoolSize(uint64(m.Size))
 	m.pool = make(chan bool, m.Size)
 	opts.SetMaxConnIdleTime(2 * time.Hour)
 	m.Ctx, _ = context.WithTimeout(context.Background(), 99999*time.Hour)