feat:优化

data_analysis/config.json

@@ -2,11 +2,9 @@
 	"updateBathSize": 100,
 	"insertBathSize": 200,
 	"selectBathSize": 200,
-	"syncPool":5,
+	"syncPool":1,
 	"updatePool":5,
 	"duration": 120,
 	"jianyu":["dataexport_order","refund_record","return_money_record","contract","nps_questions","nps_words","invoice"],
-	"bxt":["dataexport_order"],
-	"jypoints":["point_type"],
-	"jyactivities":[]
+	"timeoutWarn":"http://172.17.4.195:19281/_send/_mail?program=data_analysis&to=wangchuanjin@topnet.net.cn&title=data_analysis程序报警&body=%s"
 }

data_analysis/config/config.go

@@ -10,21 +10,16 @@ type config struct {
 	UpdatePool     int
 	Duration       int
 	Jianyu         []string
-	Bxt            []string
-	Jypoints       []string
-	Jyactivities   []string
+	TimeoutWarn    string
 }
 type timeTask struct {
-	Datetime           string `json:"datetime"`
-	Action_login_id    int64  `json:"action_login_id"`
-	Raw_user_id        int64  `json:"raw_user_id"`
-	Bxt_raw_user_id    int64  `json:"bxt_raw_user_id"`
-	Integral_flow_id   int64  `json:"integral_flow_id"`
-	User_prize_id      int64  `json:"user_prize_id"`
-	Weixin_pay_id      int64  `json:"weixin_pay_id"`
-	Ali_pay_id         int64  `json:"ali_pay_id"`
-	Nps_user_id        int64  `json:"nps_user_id"`
-	MoneyCorrection_id int64  `json:"moneyCorrection_id"`
+	Datetime            string `json:"datetime"`
+	Raw_user_id         int64  `json:"raw_user_id"`
+	Weixin_pay_id       int64  `json:"weixin_pay_id"`
+	Ali_pay_id          int64  `json:"ali_pay_id"`
+	Nps_user_id         int64  `json:"nps_user_id"`
+	MoneyCorrection_id  int64  `json:"moneyCorrection_id"`
+	Message_send_log_id int64  `json:"message_send_log_id"`
 }
 
 var (

data_analysis/db.json

@@ -1,76 +1,12 @@
 {
-	"mongodb": {
-		"main": {
-			"address": "192.168.3.128:27080",
-	 		"size": 5,
-	 		"dbName": "qfw",
-			"replSet": ""
-		}
-	},
     "mysql": {
-	    "main_jianyu": {
-	        "dbName": "jianyu",
-	        "address": "192.168.3.166:3306",
-	        "userName": "root",
-	        "passWord": "Topnet@123",
-			"maxOpenConns": 5,
-			"maxIdleConns": 5
-	    },
-		"main_bxt": {
-	        "dbName": "bxt",
-	        "address": "192.168.3.166:3306",
-	        "userName": "root",
-	        "passWord": "Topnet@123",
-			"maxOpenConns": 5,
-			"maxIdleConns": 5
-	    },
-		"to_jianyu": {
-	        "dbName": "wcj",
-	        "address": "192.168.3.242:3306",
-	        "userName": "root",
-	        "passWord": "Topnet123",
-			"maxOpenConns": 5,
-			"maxIdleConns": 5
-	    },
-		"to_bxt": {
-	        "dbName": "bxt",
-	        "address": "192.168.3.71:3306",
-	        "userName": "root",
-	        "passWord": "Topnet123",
-			"maxOpenConns": 5,
-			"maxIdleConns": 5
-	    },
-		"from_jianyu": {
-	        "dbName": "bxt",
-	        "address": "192.168.3.242:3306",
-	        "userName": "root",
-	        "passWord": "Topnet123",
-			"maxOpenConns": 5,
-			"maxIdleConns": 5
-	    },
-		"from_bxt": {
-	        "dbName": "bxt",
-	        "address": "192.168.3.11:3366",
-	        "userName": "root",
-	        "passWord": "Topnet123",
-			"maxOpenConns": 5,
-			"maxIdleConns": 5
-	    },
-		"from_jyactivites": {
-	        "dbName": "jyactivites",
-	        "address": "192.168.3.11:3366",
-	        "userName": "root",
-	        "passWord": "Topnet123",
-			"maxOpenConns": 5,
-			"maxIdleConns": 5
-	    },
-		"from_jypoints": {
-	        "dbName": "jypoints",
-	        "address": "192.168.3.11:3366",
-	        "userName": "root",
-	        "passWord": "Topnet123",
-			"maxOpenConns": 5,
-			"maxIdleConns": 5
+		"main": {
+	        "dbName": "data_analysis",
+            "address": "192.168.3.217:4000",
+            "userName": "root",
+            "passWord": "=PDT49#80Z!RVv52_z",
+            "maxOpenConns": 5,
+            "maxIdleConns": 5
 	    }
     }
 }

data_analysis/db/db.go

@@ -4,21 +4,12 @@ import (
 	"log"
 
 	util "app.yhyue.com/moapp/jybase/common"
-	. "app.yhyue.com/moapp/jybase/mongodb"
 	"app.yhyue.com/moapp/jybase/mysql"
 )
 
 var (
-	DbConf                  *dbConf
-	Mysql_Main_Jianyu       *mysql.Mysql
-	Mysql_Main_Bxt          *mysql.Mysql
-	Mysql_To_Jianyu         *mysql.Mysql
-	Mysql_From_Jianyu       *mysql.Mysql
-	Mysql_To_Bxt            *mysql.Mysql
-	Mysql_From_Bxt          *mysql.Mysql
-	Mysql_From_Jyactivities *mysql.Mysql
-	Mysql_From_Jypoints     *mysql.Mysql
-	Mgo                     MongodbSim
+	DbConf     *dbConf
+	Mysql_Main *mysql.Mysql
 )
 
 type dbConf struct {
@@ -26,14 +17,7 @@ type dbConf struct {
 		Main *mgoConf
 	}
 	Mysql struct {
-		Main_jianyu       *mysqlConf
-		Main_bxt          *mysqlConf
-		To_jianyu         *mysqlConf
-		From_jianyu       *mysqlConf
-		To_bxt            *mysqlConf
-		From_bxt          *mysqlConf
-		From_jyactivities *mysqlConf
-		From_jypoints     *mysqlConf
+		Main *mysqlConf
 	}
 }
 type mgoConf struct {
@@ -57,111 +41,17 @@ type mysqlConf struct {
 func init() {
 	util.ReadConfig("./db.json", &DbConf)
 	if DbConf != nil {
-		if DbConf.Mongodb.Main != nil {
-			log.Println("初始化 mongodb main")
-			Mgo = MongodbSim{
-				MongodbAddr: DbConf.Mongodb.Main.Address,
-				Size:        DbConf.Mongodb.Main.Size,
-				DbName:      DbConf.Mongodb.Main.DbName,
-				ReplSet:     DbConf.Mongodb.Main.ReplSet,
-			}
-			Mgo.InitPool()
-		}
-		if DbConf.Mysql.Main_jianyu != nil {
-			log.Println("初始化 mysql main jianyu")
-			Mysql_Main_Jianyu = &mysql.Mysql{
-				Address:      DbConf.Mysql.Main_jianyu.Address,
-				UserName:     DbConf.Mysql.Main_jianyu.UserName,
-				PassWord:     DbConf.Mysql.Main_jianyu.PassWord,
-				DBName:       DbConf.Mysql.Main_jianyu.DbName,
-				MaxOpenConns: DbConf.Mysql.Main_jianyu.MaxOpenConns,
-				MaxIdleConns: DbConf.Mysql.Main_jianyu.MaxIdleConns,
-			}
-			Mysql_Main_Jianyu.Init()
-		}
-		if DbConf.Mysql.Main_bxt != nil {
-			log.Println("初始化 mysql main bxt")
-			Mysql_Main_Bxt = &mysql.Mysql{
-				Address:      DbConf.Mysql.Main_bxt.Address,
-				UserName:     DbConf.Mysql.Main_bxt.UserName,
-				PassWord:     DbConf.Mysql.Main_bxt.PassWord,
-				DBName:       DbConf.Mysql.Main_bxt.DbName,
-				MaxOpenConns: DbConf.Mysql.Main_bxt.MaxOpenConns,
-				MaxIdleConns: DbConf.Mysql.Main_bxt.MaxIdleConns,
-			}
-			Mysql_Main_Bxt.Init()
-		}
-		if DbConf.Mysql.To_jianyu != nil {
-			log.Println("初始化 mysql to jianyu")
-			Mysql_To_Jianyu = &mysql.Mysql{
-				Address:      DbConf.Mysql.To_jianyu.Address,
-				UserName:     DbConf.Mysql.To_jianyu.UserName,
-				PassWord:     DbConf.Mysql.To_jianyu.PassWord,
-				DBName:       DbConf.Mysql.To_jianyu.DbName,
-				MaxOpenConns: DbConf.Mysql.To_jianyu.MaxOpenConns,
-				MaxIdleConns: DbConf.Mysql.To_jianyu.MaxIdleConns,
-			}
-			Mysql_To_Jianyu.Init()
-		}
-		if DbConf.Mysql.To_bxt != nil {
-			log.Println("初始化 mysql to bxt")
-			Mysql_To_Bxt = &mysql.Mysql{
-				Address:      DbConf.Mysql.To_bxt.Address,
-				UserName:     DbConf.Mysql.To_bxt.UserName,
-				PassWord:     DbConf.Mysql.To_bxt.PassWord,
-				DBName:       DbConf.Mysql.To_bxt.DbName,
-				MaxOpenConns: DbConf.Mysql.To_bxt.MaxOpenConns,
-				MaxIdleConns: DbConf.Mysql.To_bxt.MaxIdleConns,
-			}
-			Mysql_To_Bxt.Init()
-		}
-		if DbConf.Mysql.From_jianyu != nil {
-			log.Println("初始化 mysql from jianyu")
-			Mysql_From_Jianyu = &mysql.Mysql{
-				Address:      DbConf.Mysql.From_jianyu.Address,
-				UserName:     DbConf.Mysql.From_jianyu.UserName,
-				PassWord:     DbConf.Mysql.From_jianyu.PassWord,
-				DBName:       DbConf.Mysql.From_jianyu.DbName,
-				MaxOpenConns: DbConf.Mysql.From_jianyu.MaxOpenConns,
-				MaxIdleConns: DbConf.Mysql.From_jianyu.MaxIdleConns,
-			}
-			Mysql_From_Jianyu.Init()
-		}
-		if DbConf.Mysql.From_bxt != nil {
-			log.Println("初始化 mysql from bxt")
-			Mysql_From_Bxt = &mysql.Mysql{
-				Address:      DbConf.Mysql.From_bxt.Address,
-				UserName:     DbConf.Mysql.From_bxt.UserName,
-				PassWord:     DbConf.Mysql.From_bxt.PassWord,
-				DBName:       DbConf.Mysql.From_bxt.DbName,
-				MaxOpenConns: DbConf.Mysql.From_bxt.MaxOpenConns,
-				MaxIdleConns: DbConf.Mysql.From_bxt.MaxIdleConns,
-			}
-			Mysql_From_Bxt.Init()
-		}
-		if DbConf.Mysql.From_jyactivities != nil {
-			log.Println("初始化 mysql from jyactivities")
-			Mysql_From_Jyactivities = &mysql.Mysql{
-				Address:      DbConf.Mysql.From_jyactivities.Address,
-				UserName:     DbConf.Mysql.From_jyactivities.UserName,
-				PassWord:     DbConf.Mysql.From_jyactivities.PassWord,
-				DBName:       DbConf.Mysql.From_jyactivities.DbName,
-				MaxOpenConns: DbConf.Mysql.From_jyactivities.MaxOpenConns,
-				MaxIdleConns: DbConf.Mysql.From_jyactivities.MaxIdleConns,
-			}
-			Mysql_From_Jyactivities.Init()
-		}
-		if DbConf.Mysql.From_jypoints != nil {
-			log.Println("初始化 mysql from jypoints")
-			Mysql_From_Jypoints = &mysql.Mysql{
-				Address:      DbConf.Mysql.From_jypoints.Address,
-				UserName:     DbConf.Mysql.From_jypoints.UserName,
-				PassWord:     DbConf.Mysql.From_jypoints.PassWord,
-				DBName:       DbConf.Mysql.From_jypoints.DbName,
-				MaxOpenConns: DbConf.Mysql.From_jypoints.MaxOpenConns,
-				MaxIdleConns: DbConf.Mysql.From_jypoints.MaxIdleConns,
-			}
-			Mysql_From_Jypoints.Init()
+		if DbConf.Mysql.Main != nil {
+			log.Println("初始化 mysql main")
+			Mysql_Main = &mysql.Mysql{
+				Address:      DbConf.Mysql.Main.Address,
+				UserName:     DbConf.Mysql.Main.UserName,
+				PassWord:     DbConf.Mysql.Main.PassWord,
+				DBName:       DbConf.Mysql.Main.DbName,
+				MaxOpenConns: DbConf.Mysql.Main.MaxOpenConns,
+				MaxIdleConns: DbConf.Mysql.Main.MaxIdleConns,
+			}
+			Mysql_Main.Init()
 		}
 	}
 }

data_analysis/entity/action_login.go

@@ -1,27 +0,0 @@
-package entity
-
-import (
-	. "data_analysis/config"
-	. "data_analysis/db"
-)
-
-var (
-	Action_login *action_login
-)
-
-type action_login struct {
-}
-
-func (a *action_login) TableName() string {
-	return "action_login"
-}
-
-//
-func (a *action_login) SaveFields() []string {
-	return []string{"id", "login_time", "ip", "platform", "province", "city", "device", "browser", "version", "sessionId", "domain", "user_id"}
-}
-
-//
-func (a *action_login) Run(start_layout, end_layout string) {
-	TimeTask.Action_login_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), a.SaveFields(), TimeTask.Action_login_id, nil)
-}

data_analysis/entity/ali_pay.go

@@ -2,7 +2,6 @@ package entity
 
 import (
 	. "data_analysis/config"
-	. "data_analysis/db"
 )
 
 var (
@@ -13,7 +12,7 @@ type ali_pay struct {
 }
 
 func (a *ali_pay) TableName() string {
-	return "ali_pay"
+	return "jianyu.ali_pay"
 }
 
 //
@@ -23,5 +22,5 @@ func (a *ali_pay) SaveFields() []string {
 
 //
 func (a *ali_pay) Run(start_layout, end_layout string) {
-	TimeTask.Ali_pay_id = sync_add(Mysql_From_Jianyu, Mysql_To_Jianyu, a.TableName(), a.SaveFields(), TimeTask.Ali_pay_id, nil)
+	TimeTask.Ali_pay_id = sync_add(a.TableName(), a.SaveFields(), TimeTask.Ali_pay_id, nil)
 }

data_analysis/entity/bxt.go

@@ -1,23 +0,0 @@
-package entity
-
-import (
-	. "data_analysis/db"
-)
-
-type Bxt struct {
-	Name string
-}
-
-func (b *Bxt) TableName() string {
-	return b.Name
-}
-
-//
-func (b *Bxt) SaveFields() []string {
-	return nil
-}
-
-//
-func (b *Bxt) Run(start_layout, end_layout string) {
-	sync_full(b, Mysql_From_Bxt, Mysql_To_Bxt)
-}

data_analysis/entity/bxt_raw_user.go

@@ -1,30 +0,0 @@
-package entity
-
-import (
-	. "data_analysis/config"
-	. "data_analysis/db"
-)
-
-var (
-	Bxt_raw_user *bxt_raw_user
-)
-
-type bxt_raw_user struct {
-}
-
-func (b *bxt_raw_user) TableName() string {
-	return "raw_user"
-}
-
-//
-func (b *bxt_raw_user) SaveFields() []string {
-	return []string{"id", "user_id", "reg_time", "province", "city", "reg_type", "device", "company", "job", "source_module", "source_channel", "follow_status", "phone", "openid", "channel_id", "name", "email", "timestamp"}
-}
-
-//
-func (b *bxt_raw_user) Run(start_layout, end_layout string) {
-	if start_layout != "" {
-		sync_update(Mysql_Main_Jianyu, Mysql_To_Jianyu, b.TableName(), start_layout, end_layout, b.SaveFields(), TimeTask.Bxt_raw_user_id, "timestamp")
-	}
-	TimeTask.Bxt_raw_user_id = sync_add(Mysql_Main_Bxt, Mysql_To_Bxt, b.TableName(), b.SaveFields(), TimeTask.Bxt_raw_user_id, nil)
-}

data_analysis/entity/entity.go

@@ -2,13 +2,18 @@ package entity
 
 import (
 	. "data_analysis/config"
+	. "data_analysis/db"
 	"fmt"
 	"log"
+	"net/http"
 	"strings"
 	"time"
 
 	. "app.yhyue.com/moapp/jybase/common"
-	"app.yhyue.com/moapp/jybase/mysql"
+)
+
+const (
+	ToDbName = "data_analysis"
 )
 
 type Entity interface {
@@ -18,18 +23,19 @@ type Entity interface {
 }
 
 //
-func sync_add(main, to *mysql.Mysql, tableName string, saveField []string, id int64, f func(fv map[string]interface{})) (lastId int64) {
-	log.Println("开始同步", tableName, "表。。。")
+func sync_add(fromTable string, saveField []string, id int64, f func(fv map[string]interface{})) (lastId int64) {
+	log.Println("开始同步", fromTable, "表。。。")
+	toTable := GetToTable(fromTable)
 	index := 0
 	array := []interface{}{}
 	lastId = id
 	q := ""
 	if id < 0 {
-		q = fmt.Sprintf(`select * from %s`, tableName)
+		q = fmt.Sprintf(`select * from %s`, fromTable)
 	} else {
-		q = fmt.Sprintf(`select * from %s where id>%d order by id`, tableName, id)
+		q = fmt.Sprintf(`select * from %s where id>%d order by id`, fromTable, id)
 	}
-	main.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
+	Mysql_Main.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
 		for _, v := range *l {
 			index++
 			lastId = Int64All(v["id"])
@@ -40,26 +46,27 @@ func sync_add(main, to *mysql.Mysql, tableName string, saveField []string, id in
 				array = append(array, v[field])
 			}
 			if index%Config.InsertBathSize == 0 {
-				log.Println("同步", tableName, "表", index)
-				to.InsertIgnoreBatch(tableName, saveField, array)
+				log.Println("同步", fromTable, "表", index)
+				Mysql_Main.InsertBatch(toTable, saveField, array)
 				array = []interface{}{}
 			}
 		}
 	}, q)
 	if len(array) > 0 {
-		to.InsertIgnoreBatch(tableName, saveField, array)
+		Mysql_Main.InsertBatch(toTable, saveField, array)
 		array = []interface{}{}
 	}
-	log.Println("同步", tableName, "表结束。。。", index)
+	log.Println("同步", fromTable, "表结束。。。", index)
 	return
 }
 
 //
-func sync_update(main, to *mysql.Mysql, tableName, start_layout, end_layout string, saveFields []string, id int64, updateTimeField string) {
-	log.Println("开始同步更新", tableName, "表 。。。")
+func sync_update(fromTable, start_layout, end_layout string, saveFields []string, id int64, updateTimeField string) {
+	toTable := GetToTable(fromTable)
+	log.Println("开始同步更新", toTable, "表 。。。")
 	index := 0
 	array := [][]interface{}{}
-	main.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
+	Mysql_Main.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
 		for _, v := range *l {
 			index++
 			datas := []interface{}{}
@@ -68,68 +75,90 @@ func sync_update(main, to *mysql.Mysql, tableName, start_layout, end_layout stri
 			}
 			array = append(array, datas)
 			if index%Config.UpdateBathSize == 0 {
-				log.Println("同步更新", tableName, "表", index)
-				to.UpdateBath(tableName, saveFields, array)
+				log.Println("同步更新", toTable, "表", index)
+				Mysql_Main.UpdateBath(toTable, saveFields, array)
 				array = [][]interface{}{}
 			}
 		}
-	}, fmt.Sprintf(`select * from %s where id<? and `+updateTimeField+`>=? and `+updateTimeField+`<?`, tableName), id, start_layout, end_layout)
+	}, fmt.Sprintf(`select * from %s where id<? and `+updateTimeField+`>=? and `+updateTimeField+`<?`, fromTable), id, start_layout, end_layout)
 	if len(array) > 0 {
-		to.UpdateBath(tableName, saveFields, array)
+		Mysql_Main.UpdateBath(toTable, saveFields, array)
 		array = [][]interface{}{}
 	}
-	log.Println("同步更新", tableName, "表结束。。。", index)
+	log.Println("同步更新", toTable, "表结束。。。", index)
 }
 
 //全量同步
-func sync_full(e Entity, from, to *mysql.Mysql) bool {
-	if to.Address != "172.17.145.164:14000" || to.DBName != "data_analysis" {
-		log.Println("数据库配置错误")
+func sync_full(e Entity) bool {
+	if Mysql_Main.Address != "172.17.145.164:14000" || Mysql_Main.DBName != ToDbName {
+		log.Fatalln("数据库配置错误")
 		return false
 	}
-	list := from.SelectBySql(`SHOW CREATE TABLE ` + e.TableName())
+	toTable := GetToTable(e.TableName())
+	list := Mysql_Main.SelectBySql(`SHOW CREATE TABLE ` + e.TableName())
 	if list != nil && len(*list) == 1 {
-		if _, err := to.DB.Exec(`drop table ` + e.TableName()); err != nil {
-			log.Println(e.TableName(), "清空表出错", err)
+		if _, err := Mysql_Main.DB.Exec(`drop table ` + toTable); err != nil {
+			log.Println(toTable, "清空表出错", err)
 			return false
 		} else {
-			log.Println(e.TableName(), "清空表完成")
+			log.Println(toTable, "清空表完成")
 		}
 		createTable := (*list)[0]["Create Table"].(string)
-		log.Println(e.TableName(), "建表语句", createTable)
-		if _, err := to.DB.Exec(createTable); err != nil {
-			log.Println(e.TableName(), "创建表错误", err)
+		log.Println(toTable, "建表语句", createTable)
+		if _, err := Mysql_Main.DB.Exec(createTable); err != nil {
+			log.Println(toTable, "创建表错误", err)
 			return false
 		} else {
-			log.Println(e.TableName(), "表创建成功")
+			log.Println(toTable, "表创建成功")
 		}
 	} else {
 		log.Println("没有", e.TableName(), "的建表语句")
 		return false
 	}
-	array := from.SelectBySql(`select id from ` + e.TableName() + ` order by id desc limit 1`)
+	array := Mysql_Main.SelectBySql(`select id from ` + e.TableName() + ` order by id desc limit 1`)
 	if array == nil || len(*array) == 0 {
 		log.Println(e.TableName(), "表没有数据")
 		return false
 	}
 	lastId := (*array)[0]["id"].(int64)
-	fields := getFields(from, e.TableName())
+	fields := getFields(strings.Split(e.TableName(), ".")[0], strings.Split(e.TableName(), ".")[1])
+	var startId int64 = -1
+	sleepMin := 1
 	for {
-		id := sync_add(from, to, e.TableName(), fields, -1, nil)
+		id := sync_add(e.TableName(), fields, startId, nil)
 		if id >= lastId {
 			break
 		}
-		log.Println("同步", e.TableName(), "数量不够，一分钟后重试")
-		time.Sleep(time.Minute)
+		startId = id
+		if sleepMin > 3 {
+			SendMail("重试5次，全量同步" + e.TableName() + "失败，请排查")
+			break
+		}
+		slm := time.Duration(sleepMin) * 2 * time.Minute
+		SendMail("全量同步" + e.TableName() + "数量不够，" + fmt.Sprint(slm.Minutes()) + "分钟后重试")
+		time.Sleep(slm)
+		sleepMin++
 	}
 	return true
 }
 
-func getFields(m *mysql.Mysql, tableName string) []string {
-	list := m.SelectBySql(`select group_concat(COLUMN_NAME separator ',') as fields from information_schema.COLUMNS where table_schema=? and table_name=?`, m.DBName, tableName)
+func getFields(dbName, tableName string) []string {
+	list := Mysql_Main.SelectBySql(`select group_concat(COLUMN_NAME separator ',') as fields from information_schema.COLUMNS where table_schema=? and table_name=?`, dbName, tableName)
 	if list == nil || len(*list) == 0 || (*list)[0]["fields"] == nil {
-		log.Println("没有找到", m.DBName, "库", tableName, "表的字段")
+		log.Println("没有找到", dbName, "库", tableName, "表的字段")
 		return nil
 	}
 	return strings.Split((*list)[0]["fields"].(string), ",")
 }
+
+func SendMail(msg string) {
+	log.Println(msg)
+	if _, err := http.Get(fmt.Sprintf(Config.TimeoutWarn, msg)); err != nil {
+		log.Println(msg, "发送告警邮件异常", err)
+	}
+}
+
+//
+func GetToTable(fromTable string) string {
+	return ToDbName + "." + strings.Split(fromTable, ".")[1]
+}

data_analysis/entity/finaljob.go

@@ -2,6 +2,7 @@ package entity
 
 import (
 	. "data_analysis/config"
+	. "data_analysis/db"
 	"database/sql"
 	"fmt"
 	"log"
@@ -10,18 +11,13 @@ import (
 	"sync/atomic"
 	"time"
 
-	"app.yhyue.com/moapp/jybase/mysql"
-
 	. "app.yhyue.com/moapp/jybase/common"
 )
 
-func NewFinalJob(tableName string, from, to *mysql.Mysql) *FinalJob {
+func NewFinalJob(tableName string) *FinalJob {
 	f := &FinalJob{
 		Name: tableName,
-		From: from,
-		To:   to,
 	}
-	f.LoadDatas()
 	return f
 }
 
@@ -29,8 +25,6 @@ type FinalJob struct {
 	Name   string
 	Fields []string
 	All    map[int64]string
-	From   *mysql.Mysql
-	To     *mysql.Mysql
 }
 
 func (f *FinalJob) TableName() string {
@@ -43,70 +37,83 @@ func (f *FinalJob) SaveFields() []string {
 }
 
 //
-func (f *FinalJob) LoadDatas() {
-	log.Println("开始加载内网", f.TableName(), "数据。。。")
-	f.Fields = getFields(f.To, f.TableName())
+func (f *FinalJob) LoadDatas(toTable string) {
+	log.Println("开始加载内网", toTable, "数据。。。")
+	f.Fields = getFields(strings.Split(toTable, ".")[0], strings.Split(toTable, ".")[1])
 	if f.Fields == nil {
 		return
 	}
 	index := 0
-	count := f.To.CountBySql(`select count(1) as count from ` + f.TableName())
-	log.Println("获取内网", f.TableName(), "表总数", count)
+	count := Mysql_Main.CountBySql(`select count(1) as count from ` + toTable)
+	log.Println("获取内网", toTable, "表总数", count)
 	if count < 0 {
 		return
 	}
+	var lastId int64
+	f.All = map[int64]string{}
+	reTry := 0
 	for {
-		f.All = map[int64]string{}
-		f.To.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
+		Mysql_Main.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
 			for _, v := range *l {
 				index++
 				var result strings.Builder
 				for _, field := range f.Fields {
 					result.WriteString(fmt.Sprintf("%v", v[field]))
 				}
-				f.All[Int64All(v["id"])] = GetMd5String(result.String())
+				lastId = Int64All(v["id"])
+				f.All[lastId] = GetMd5String(result.String())
 				if index%500 == 0 {
-					log.Println("加载内网", f.TableName(), "数据", index)
+					log.Println("加载内网", toTable, "数据", index)
 				}
 			}
-		}, `select * from `+f.TableName())
+		}, `select * from `+toTable+` where id>?`, lastId)
 		if int64(len(f.All)) != count {
-			log.Println("加载内网", f.TableName(), "数据不够", len(f.All), count, "休眠30秒重试")
-			time.Sleep(30 * time.Second)
+			reTry++
+			if reTry == 3 {
+				log.Fatalln("加载内网", toTable, "数据失败", len(f.All), count, "请检查")
+			}
+			slm := time.Duration(reTry) * 2 * time.Minute
+			SendMail(fmt.Sprintf("加载内网%s数据不够%d<>%d，休眠%f分钟重试", toTable, len(f.All), count, slm.Minutes()))
+			time.Sleep(slm)
 		} else {
 			break
 		}
 	}
-	log.Println("加载内网", f.TableName(), "数据结束。。。", index)
+	log.Println("加载内网", toTable, "数据结束。。。", index)
 }
 
 //
 func (f *FinalJob) Run(start_layout, end_layout string) {
-	if fromFields, toFields := getFields(f.From, f.TableName()), getFields(f.To, f.TableName()); strings.Join(fromFields, ",") != strings.Join(toFields, ",") {
+	fromDbName, fromTableName := strings.Split(f.TableName(), ".")[0], strings.Split(f.TableName(), ".")[1]
+	toTable := GetToTable(f.TableName())
+	if fromFields, toFields := getFields(fromDbName, fromTableName), getFields(ToDbName, fromTableName); strings.Join(fromFields, ",") != strings.Join(toFields, ",") {
 		if fromFields == nil || toFields == nil {
 			return
 		}
-		log.Println(f.TableName(), "内网字段和公网字段不一致，全量同步", toFields, fromFields)
-		sync_full(f, f.From, f.To)
-		f.LoadDatas()
+		log.Println(fromTableName, "内网字段和公网字段不一致，全量同步", toFields, fromFields)
+		sync_full(f)
+		f.LoadDatas(toTable)
 		return
 	}
-	fromCount := f.From.CountBySql(`select count(1) as count from ` + f.TableName())
-	toCount := f.To.CountBySql(`select count(1) as count from ` + f.TableName())
+	fromCount := Mysql_Main.CountBySql(`select count(1) as count from ` + f.TableName())
+	toCount := Mysql_Main.CountBySql(`select count(1) as count from ` + toTable)
 	if toCount > fromCount {
-		log.Println(f.TableName(), "内网数量大于公网数量，全量同步", toCount, fromCount)
-		sync_full(f, f.From, f.To)
-		f.LoadDatas()
+		log.Println(fromTableName, "内网数量大于公网数量，全量同步", toCount, fromCount)
+		sync_full(f)
+		f.LoadDatas(toTable)
 		return
 	}
-	log.Println("开始对比公网", f.TableName(), "数据。。。")
+	log.Println("开始对比公网", fromTableName, "数据。。。")
+	if f.All == nil || len(f.All) == 0 {
+		f.LoadDatas(toTable)
+	}
 	index := 0
 	var addIndex int64
 	var updateIndex int64
 	pool := make(chan bool, Config.UpdatePool)
 	wait := &sync.WaitGroup{}
 	lock := &sync.Mutex{}
-	f.From.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
+	Mysql_Main.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
 		for _, v := range *l {
 			index++
 			var result strings.Builder
@@ -128,7 +135,7 @@ func (f *FinalJob) Run(start_layout, end_layout string) {
 				allVal := f.All[id]
 				lock.Unlock()
 				if allVal == "" {
-					r1, r2 := f.To.InsertBatch(f.TableName(), f.SaveFields(), delInsertArr)
+					r1, r2 := Mysql_Main.InsertBatch(toTable, f.SaveFields(), delInsertArr)
 					if r1 > 0 && r2 > 0 {
 						lock.Lock()
 						f.All[id] = resultMd5
@@ -136,9 +143,9 @@ func (f *FinalJob) Run(start_layout, end_layout string) {
 						atomic.AddInt64(&addIndex, 1)
 					}
 				} else if allVal != resultMd5 {
-					if f.To.ExecTx("", func(tx *sql.Tx) bool {
-						r3 := f.To.UpdateOrDeleteBySqlByTx(tx, `delete from `+f.TableName()+" where id=?", delInsertId)
-						r1, r2 := f.To.InsertBatchByTx(tx, f.TableName(), f.SaveFields(), delInsertArr)
+					if Mysql_Main.ExecTx("", func(tx *sql.Tx) bool {
+						r3 := Mysql_Main.UpdateOrDeleteBySqlByTx(tx, `delete from `+toTable+" where id=?", delInsertId)
+						r1, r2 := Mysql_Main.InsertBatchByTx(tx, toTable, f.SaveFields(), delInsertArr)
 						return r1 > 0 && r2 > 0 && r3 > 0
 					}) {
 						lock.Lock()
@@ -149,10 +156,10 @@ func (f *FinalJob) Run(start_layout, end_layout string) {
 				}
 			}(id, array)
 			if index%500 == 0 {
-				log.Println("对比公网", f.TableName(), "数据结束", index)
+				log.Println("对比公网", fromTableName, "数据结束", index)
 			}
 		}
 	}, `select `+strings.Join(f.SaveFields(), ",")+` from `+f.TableName())
 	wait.Wait()
-	log.Println("开始对比公网", f.TableName(), "数据结束。。。", index, "新增", addIndex, "更新", updateIndex)
+	log.Println("开始对比公网", fromTableName, "数据结束。。。", index, "新增", addIndex, "更新", updateIndex)
 }

data_analysis/entity/integral_flow.go

@@ -1,27 +0,0 @@
-package entity
-
-import (
-	. "data_analysis/config"
-	. "data_analysis/db"
-)
-
-var (
-	Integral_flow *integral_flow
-)
-
-type integral_flow struct {
-}
-
-func (i *integral_flow) TableName() string {
-	return "integral_flow"
-}
-
-//
-func (i *integral_flow) SaveFields() []string {
-	return []string{"id", "userId", "pointType", "SourceId", "sourceType", "point", "createTime", "endDate", "appId", "sort", "abstract", "operationType", "serialNumber"}
-}
-
-//
-func (i *integral_flow) Run(start_layout, end_layout string) {
-	TimeTask.Integral_flow_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, i.TableName(), i.SaveFields(), TimeTask.Integral_flow_id, nil)
-}

data_analysis/entity/message_send_log.go

@@ -0,0 +1,29 @@
+package entity
+
+import (
+	. "data_analysis/config"
+)
+
+var (
+	Message_send_log *message_send_log
+)
+
+type message_send_log struct {
+}
+
+func (m *message_send_log) TableName() string {
+	return "jianyu.message_send_log"
+}
+
+//
+func (m *message_send_log) SaveFields() []string {
+	return []string{"id", "send_usergroup_id", "send_usergroup_name", "receive_user_id", "msg_type", "title", "content", "send_mode", "send_time", "send_status", "update_time", "createtime", "link", "isdel", "send_userid", "send_name", "cite_id", "sign", "user_add_way", "template_name", "update_user", "show_buoy", "show_content", "menu_name"}
+}
+
+//
+func (m *message_send_log) Run(start_layout, end_layout string) {
+	if start_layout != "" {
+		sync_update(m.TableName(), start_layout, end_layout, m.SaveFields(), TimeTask.Message_send_log_id, "update_time")
+	}
+	TimeTask.Message_send_log_id = sync_add(m.TableName(), m.SaveFields(), TimeTask.Message_send_log_id, nil)
+}

data_analysis/entity/moneyCorrection.go

@@ -2,7 +2,6 @@ package entity
 
 import (
 	. "data_analysis/config"
-	. "data_analysis/db"
 )
 
 var (
@@ -13,15 +12,15 @@ type moneyCorrection struct {
 }
 
 func (m *moneyCorrection) TableName() string {
-	return "moneyCorrection"
+	return "jianyu.moneyCorrection"
 }
 
 //
 func (m *moneyCorrection) SaveFields() []string {
-	return []string{"id", "orderCode", "createTime", "orderMoney", "payMoney", "bigSubMoney", "supplyMoney", "commission", "proceduresMoney", "correctionReason", "createPerson"}
+	return []string{"id", "orderCode", "createTime", "redType", "signingSubject", "orderMoney", "payMoney", "bigSubMoney", "supplyMoney", "commission", "proceduresMoney", "correctionReason", "createPerson"}
 }
 
 //
 func (m *moneyCorrection) Run(start_layout, end_layout string) {
-	TimeTask.MoneyCorrection_id = sync_add(Mysql_From_Jianyu, Mysql_To_Jianyu, m.TableName(), m.SaveFields(), TimeTask.MoneyCorrection_id, nil)
+	TimeTask.MoneyCorrection_id = sync_add(m.TableName(), m.SaveFields(), TimeTask.MoneyCorrection_id, nil)
 }

data_analysis/entity/nps_user.go

@@ -2,7 +2,6 @@ package entity
 
 import (
 	. "data_analysis/config"
-	. "data_analysis/db"
 )
 
 var (
@@ -13,7 +12,7 @@ type nps_user struct {
 }
 
 func (n *nps_user) TableName() string {
-	return "nps_user"
+	return "jianyu.nps_user"
 }
 
 //
@@ -24,7 +23,7 @@ func (n *nps_user) SaveFields() []string {
 //
 func (n *nps_user) Run(start_layout, end_layout string) {
 	if start_layout != "" {
-		sync_update(Mysql_From_Jianyu, Mysql_To_Jianyu, n.TableName(), start_layout, end_layout, n.SaveFields(), TimeTask.Nps_user_id, "updatetime")
+		sync_update(n.TableName(), start_layout, end_layout, n.SaveFields(), TimeTask.Nps_user_id, "updatetime")
 	}
-	TimeTask.Nps_user_id = sync_add(Mysql_From_Jianyu, Mysql_To_Jianyu, n.TableName(), n.SaveFields(), TimeTask.Nps_user_id, nil)
+	TimeTask.Nps_user_id = sync_add(n.TableName(), n.SaveFields(), TimeTask.Nps_user_id, nil)
 }

data_analysis/entity/raw_user.go

@@ -2,7 +2,6 @@ package entity
 
 import (
 	. "data_analysis/config"
-	. "data_analysis/db"
 )
 
 var (
@@ -13,7 +12,7 @@ type raw_user struct {
 }
 
 func (r *raw_user) TableName() string {
-	return "raw_user"
+	return "buryingpoint.raw_user"
 }
 
 //
@@ -24,7 +23,7 @@ func (r *raw_user) SaveFields() []string {
 //
 func (r *raw_user) Run(start_layout, end_layout string) {
 	if start_layout != "" {
-		sync_update(Mysql_Main_Jianyu, Mysql_To_Jianyu, r.TableName(), start_layout, end_layout, r.SaveFields(), TimeTask.Raw_user_id, "timestamp")
+		sync_update(r.TableName(), start_layout, end_layout, r.SaveFields(), TimeTask.Raw_user_id, "timestamp")
 	}
-	TimeTask.Raw_user_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, r.TableName(), r.SaveFields(), TimeTask.Raw_user_id, nil)
+	TimeTask.Raw_user_id = sync_add(r.TableName(), r.SaveFields(), TimeTask.Raw_user_id, nil)
 }

data_analysis/entity/user_prize.go

@@ -1,30 +0,0 @@
-package entity
-
-import (
-	. "data_analysis/config"
-	. "data_analysis/db"
-)
-
-var (
-	User_prize *user_prize
-)
-
-type user_prize struct {
-}
-
-func (u *user_prize) TableName() string {
-	return "user_prize"
-}
-
-//
-func (u *user_prize) SaveFields() []string {
-	return []string{"id", "userId", "prizeId", "validityDates", "beginDate", "endDate", "useDate", "prizeType", "appId", "lotteryId", "createTime", "name", "full", "reduce", "userName", "userProduct", "activityId", "orderCode", "discount", "userPhone"}
-}
-
-//
-func (u *user_prize) Run(start_layout, end_layout string) {
-	if start_layout != "" {
-		sync_update(Mysql_From_Jyactivities, Mysql_To_Jianyu, u.TableName(), start_layout, end_layout, u.SaveFields(), TimeTask.User_prize_id, "useDate")
-	}
-	TimeTask.User_prize_id = sync_add(Mysql_From_Jyactivities, Mysql_To_Jianyu, u.TableName(), u.SaveFields(), TimeTask.User_prize_id, nil)
-}

data_analysis/entity/weixin_pay.go

@@ -2,7 +2,6 @@ package entity
 
 import (
 	. "data_analysis/config"
-	. "data_analysis/db"
 )
 
 var (
@@ -13,7 +12,7 @@ type weixin_pay struct {
 }
 
 func (w *weixin_pay) TableName() string {
-	return "weixin_pay"
+	return "jianyu.weixin_pay"
 }
 
 //
@@ -23,5 +22,5 @@ func (w *weixin_pay) SaveFields() []string {
 
 //
 func (w *weixin_pay) Run(start_layout, end_layout string) {
-	TimeTask.Weixin_pay_id = sync_add(Mysql_From_Jianyu, Mysql_To_Jianyu, w.TableName(), w.SaveFields(), TimeTask.Weixin_pay_id, nil)
+	TimeTask.Weixin_pay_id = sync_add(w.TableName(), w.SaveFields(), TimeTask.Weixin_pay_id, nil)
 }

data_analysis/main.go

@@ -17,26 +17,19 @@ import (
 var (
 	incSyncMap = map[string]Entity{
 		//jianyu
-		"nps_user":        Nps_user,
-		"ali_pay":         Ali_pay,
-		"weixin_pay":      Weixin_pay,
-		"moneyCorrection": MoneyCorrection,
+		Raw_user.TableName():         Raw_user,
+		Nps_user.TableName():         Nps_user,
+		Ali_pay.TableName():          Ali_pay,
+		Weixin_pay.TableName():       Weixin_pay,
+		MoneyCorrection.TableName():  MoneyCorrection,
+		Message_send_log.TableName(): Message_send_log,
 	}
 	fullSyncMap = map[string]Entity{}
 )
 
 func init() {
 	for _, v := range Config.Jianyu {
-		fullSyncMap["jianyu."+v] = NewFinalJob(v, Mysql_From_Jianyu, Mysql_To_Jianyu)
-	}
-	for _, v := range Config.Bxt {
-		fullSyncMap["bxt."+v] = NewFinalJob(v, Mysql_From_Bxt, Mysql_To_Bxt)
-	}
-	for _, v := range Config.Jypoints {
-		fullSyncMap["jypoints."+v] = NewFinalJob(v, Mysql_From_Jypoints, Mysql_To_Jianyu)
-	}
-	for _, v := range Config.Jyactivities {
-		fullSyncMap["jyactivities."+v] = NewFinalJob(v, Mysql_From_Jyactivities, Mysql_To_Jianyu)
+		fullSyncMap["jianyu."+v] = NewFinalJob("jianyu." + v)
 	}
 }
 
@@ -69,18 +62,23 @@ func main() {
 	}
 }
 
+var isRuning bool
+
 func start(inc, full []Entity) {
 	go func() {
-		if err := Mysql_To_Jianyu.DB.Ping(); err != nil {
-			log.Println("Mysql_To_Jianyu Ping", err)
-			Mysql_To_Jianyu.Init()
+		if isRuning {
+			SendMail("同步数据任务上轮未结束，请排查！")
+			return
 		}
-		if err := Mysql_To_Bxt.DB.Ping(); err != nil {
-			log.Println("Mysql_To_Bxt Ping", err)
-			Mysql_To_Bxt.Init()
+		isRuning = true
+		if err := Mysql_Main.DB.Ping(); err != nil {
+			log.Println("Mysql_Main Ping", err)
+			Mysql_Main.Init()
 		}
+		Mysql_Main.ExecBySql(`SET session max_execution_time=86400`)
 		incSyncRun(inc...)
 		fullSyncRun(full...)
+		isRuning = false
 	}()
 	now := time.Now()
 	next := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local)
@@ -91,7 +89,7 @@ func start(inc, full []Entity) {
 }
 
 func fullSyncRun(es ...Entity) {
-	log.Println("全量同步任务开始。。。", fmt.Sprintf("%+v", TimeTask))
+	log.Println("全量同步任务开始。。。")
 	run("", "", es...)
 	log.Println("全量同步任务结束。。。")
 }
@@ -110,22 +108,15 @@ func run(d1, d2 string, es ...Entity) {
 	wait := &sync.WaitGroup{}
 	pool := make(chan bool, Config.SyncPool)
 	for _, e := range es {
-		isTake := true
-		select {
-		case <-time.After(time.Hour):
-			isTake = false
-		case pool <- true:
-		}
+		pool <- true
 		wait.Add(1)
-		go func(v Entity, take bool) {
+		go func(v Entity) {
 			defer func() {
-				if take {
-					<-pool
-				}
+				<-pool
 				wait.Done()
 			}()
 			v.Run(d1, d2)
-		}(e, isTake)
+		}(e)
 	}
 	wait.Wait()
 }

data_analysis/timetask.json

@@ -1 +1 @@
-{"datetime":"2021-05-06 11:55:12","ali_pay_id":4225,"weixin_pay_id":45000,"nps_user_id":133330,"moneyCorrection_id":13}
+{"datetime":"2024-01-31 10:35:28","raw_user_id":0,"weixin_pay_id":1068537,"ali_pay_id":32708,"nps_user_id":133330,"moneyCorrection_id":30171,"message_send_log_id":217696}

online_syncto_offline/config.json

@@ -8,5 +8,6 @@
 	"jianyu":["dataexport_order","refund_record","return_money_record","contract","nps_questions","nps_words","invoice"],
 	"bxt":["dataexport_order"],
 	"jypoints":["point_type"],
-	"jyactivities":[]
+	"jyactivities":[],
+	"timeoutWarn":"http://172.17.4.195:19281/_send/_mail?program=online_syncto_offline&to=wangchuanjin@topnet.net.cn&title=online_syncto_offline程序报警&body=%s"
 }

online_syncto_offline/config/config.go

@@ -13,18 +13,20 @@ type config struct {
 	Bxt            []string
 	Jypoints       []string
 	Jyactivities   []string
+	TimeoutWarn    string
 }
 type timeTask struct {
-	Datetime           string `json:"datetime"`
-	Action_login_id    int64  `json:"action_login_id"`
-	Raw_user_id        int64  `json:"raw_user_id"`
-	Bxt_raw_user_id    int64  `json:"bxt_raw_user_id"`
-	Integral_flow_id   int64  `json:"integral_flow_id"`
-	User_prize_id      int64  `json:"user_prize_id"`
-	Weixin_pay_id      int64  `json:"weixin_pay_id"`
-	Ali_pay_id         int64  `json:"ali_pay_id"`
-	Nps_user_id        int64  `json:"nps_user_id"`
-	MoneyCorrection_id int64  `json:"moneyCorrection_id"`
+	Datetime            string `json:"datetime"`
+	Action_login_id     int64  `json:"action_login_id"`
+	Raw_user_id         int64  `json:"raw_user_id"`
+	Bxt_raw_user_id     int64  `json:"bxt_raw_user_id"`
+	Integral_flow_id    int64  `json:"integral_flow_id"`
+	User_prize_id       int64  `json:"user_prize_id"`
+	Weixin_pay_id       int64  `json:"weixin_pay_id"`
+	Ali_pay_id          int64  `json:"ali_pay_id"`
+	Nps_user_id         int64  `json:"nps_user_id"`
+	MoneyCorrection_id  int64  `json:"moneyCorrection_id"`
+	Message_send_log_id int64  `json:"message_send_log_id"`
 }
 
 var (

online_syncto_offline/entity/entity.go

@@ -3,6 +3,7 @@ package entity
 import (
 	"fmt"
 	"log"
+	"net/http"
 	. "online_syncto_offline/config"
 	"strings"
 	"time"
@@ -119,7 +120,7 @@ func sync_full(e Entity, from, to *mysql.Mysql) bool {
 		if id >= lastId {
 			break
 		}
-		log.Println("同步", e.TableName(), "数量不够，一分钟后重试")
+		SendMail("同步" + e.TableName() + "数量不够，一分钟后重试")
 		time.Sleep(time.Minute)
 	}
 	return true
@@ -133,3 +134,9 @@ func getFields(m *mysql.Mysql, tableName string) []string {
 	}
 	return strings.Split((*list)[0]["fields"].(string), ",")
 }
+func SendMail(msg string) {
+	log.Println(msg)
+	if _, err := http.Get(fmt.Sprintf(Config.TimeoutWarn, msg)); err != nil {
+		log.Println(msg, "发送告警邮件异常", err)
+	}
+}

online_syncto_offline/entity/finaljob.go

@@ -71,7 +71,7 @@ func (f *FinalJob) LoadDatas() {
 			}
 		}, `select * from `+f.TableName())
 		if int64(len(f.All)) != count {
-			log.Println("加载内网", f.TableName(), "数据不够", len(f.All), count, "休眠30秒重试")
+			SendMail(fmt.Sprintf("加载内网%s数据不够%d<>%d，休眠30秒重试", f.TableName(), len(f.All), count))
 			time.Sleep(30 * time.Second)
 		} else {
 			break

online_syncto_offline/entity/message_send_log.go

@@ -0,0 +1,30 @@
+package entity
+
+import (
+	. "online_syncto_offline/config"
+	. "online_syncto_offline/db"
+)
+
+var (
+	Message_send_log *message_send_log
+)
+
+type message_send_log struct {
+}
+
+func (m *message_send_log) TableName() string {
+	return "message_send_log"
+}
+
+//
+func (m *message_send_log) SaveFields() []string {
+	return []string{"id", "send_usergroup_id", "send_usergroup_name", "receive_user_id", "msg_type", "title", "content", "send_mode", "send_time", "send_status", "update_time", "createtime", "link", "isdel", "send_userid", "send_name", "cite_id", "send_num"}
+}
+
+//
+func (m *message_send_log) Run(start_layout, end_layout string) {
+	if start_layout != "" {
+		sync_update(Mysql_From_Jianyu, Mysql_To_Jianyu, m.TableName(), start_layout, end_layout, m.SaveFields(), TimeTask.Message_send_log_id, "update_time")
+	}
+	TimeTask.Message_send_log_id = sync_add(Mysql_From_Jianyu, Mysql_To_Jianyu, m.TableName(), m.SaveFields(), TimeTask.Message_send_log_id, nil)
+}

online_syncto_offline/entity/moneyCorrection.go

@@ -18,7 +18,7 @@ func (m *moneyCorrection) TableName() string {
 
 //
 func (m *moneyCorrection) SaveFields() []string {
-	return []string{"id", "orderCode", "createTime", "orderMoney", "payMoney", "bigSubMoney", "supplyMoney", "commission", "proceduresMoney", "correctionReason", "createPerson"}
+	return []string{"id", "orderCode", "createTime", "redType", "signingSubject", "orderMoney", "payMoney", "bigSubMoney", "supplyMoney", "commission", "proceduresMoney", "correctionReason", "createPerson"}
 }
 
 //

online_syncto_offline/main.go

@@ -76,18 +76,28 @@ func main() {
 	}
 }
 
+var isRuning bool
+
 func start(inc, full []Entity) {
 	go func() {
+		if isRuning {
+			SendMail("同步数据任务上轮未结束，请排查！")
+			return
+		}
+		isRuning = true
 		if err := Mysql_To_Jianyu.DB.Ping(); err != nil {
 			log.Println("Mysql_To_Jianyu Ping", err)
 			Mysql_To_Jianyu.Init()
 		}
+		Mysql_To_Jianyu.ExecBySql(`SET session max_execution_time=86400`)
 		if err := Mysql_To_Bxt.DB.Ping(); err != nil {
 			log.Println("Mysql_To_Bxt Ping", err)
 			Mysql_To_Bxt.Init()
 		}
+		Mysql_To_Bxt.ExecBySql(`SET session max_execution_time=86400`)
 		incSyncRun(inc...)
 		fullSyncRun(full...)
+		isRuning = false
 	}()
 	now := time.Now()
 	next := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local)
@@ -117,22 +127,15 @@ func run(d1, d2 string, es ...Entity) {
 	wait := &sync.WaitGroup{}
 	pool := make(chan bool, Config.SyncPool)
 	for _, e := range es {
-		isTake := true
-		select {
-		case <-time.After(time.Hour):
-			isTake = false
-		case pool <- true:
-		}
+		pool <- true
 		wait.Add(1)
-		go func(v Entity, take bool) {
+		go func(v Entity) {
 			defer func() {
-				if take {
-					<-pool
-				}
+				<-pool
 				wait.Done()
 			}()
 			v.Run(d1, d2)
-		}(e, isTake)
+		}(e)
 	}
 	wait.Wait()
 }

online_syncto_offline/timetask.json

@@ -1 +1 @@
-{"datetime":"2021-05-06 11:55:12","action_login_id":75312,"raw_user_id":1933484,"bxt_raw_user_id":1933484,"integral_flow_id":2601,"ali_pay_id":4225,"weixin_pay_id":45000,"user_prize_id":608701,"nps_user_id":133330,"moneyCorrection_id":13}
+{"datetime":"2021-05-06 11:55:12","action_login_id":75312,"raw_user_id":1933484,"bxt_raw_user_id":1933484,"integral_flow_id":2601,"ali_pay_id":4225,"weixin_pay_id":45000,"user_prize_id":608701,"nps_user_id":133330,"moneyCorrection_id":13,"message_send_log_id":100}