xiugai

online_datasync/config/config.go

@@ -11,13 +11,14 @@ type config struct {
 	SelectMgoUserPool int
 }
 type timeTask struct {
-	Datetime            string `json:"datetime"`
-	User_mgo_mysql_id   string `json:"user_mgo_mysql_id"`
-	Raw_product_id      int64  `json:"raw_product_id"`
-	Dataexport_order_id int64  `json:"dataexport_order_id"`
-	Point_type_id       int64  `json:"point_type_id"`
-	Integral_flow_id    int64  `json:"integral_flow_id"`
-	User_prize_id       int64  `json:"user_prize_id"`
+	Datetime              string `json:"datetime"`
+	User_mgo_mysql_id     string `json:"user_mgo_mysql_id"`
+	Bxt_user_mgo_mysql_id string `json:"bxt_user_mgo_mysql_id"`
+	Raw_product_id        int64  `json:"raw_product_id"`
+	Dataexport_order_id   int64  `json:"dataexport_order_id"`
+	Point_type_id         int64  `json:"point_type_id"`
+	Integral_flow_id      int64  `json:"integral_flow_id"`
+	User_prize_id         int64  `json:"user_prize_id"`
 }
 
 var (

online_datasync/db.json

@@ -15,6 +15,14 @@
 	        "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
 	    },
 		"from_jianyu": {
 	        "dbName": "jianyu",

online_datasync/db/db.go

@@ -11,6 +11,7 @@ import (
 var (
 	DbConf                  *dbConf
 	Mysql_Main              *mysql.Mysql
+	Mysql_Main_Bxt          *mysql.Mysql
 	Mysql_From_Jianyu       *mysql.Mysql
 	Mysql_From_Jypoints     *mysql.Mysql
 	Mysql_From_Jyactivities *mysql.Mysql
@@ -23,6 +24,7 @@ type dbConf struct {
 	}
 	Mysql struct {
 		Main              *mysqlConf
+		Main_bxt          *mysqlConf
 		From_jianyu       *mysqlConf
 		From_jyactivities *mysqlConf
 		From_jypoints     *mysqlConf
@@ -60,7 +62,7 @@ func init() {
 			Mgo.InitPool()
 		}
 		if DbConf.Mysql.Main != nil {
-			log.Println("初始化 mysql main")
+			log.Println("初始化 mysql main jianyu")
 			Mysql_Main = &mysql.Mysql{
 				Address:      DbConf.Mysql.Main.Address,
 				UserName:     DbConf.Mysql.Main.UserName,
@@ -71,6 +73,18 @@ func init() {
 			}
 			Mysql_Main.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.From_jianyu != nil {
 			log.Println("初始化 mysql from jianyu")
 			Mysql_From_Jianyu = &mysql.Mysql{

online_datasync/entity/bxt_raw_user .go

@@ -0,0 +1,192 @@
+package entity
+
+import (
+	"log"
+	. "online_datasync/config"
+	. "online_datasync/db"
+	"strings"
+	"sync"
+
+	. "app.yhyue.com/moapp/jybase/common"
+	. "app.yhyue.com/moapp/jybase/date"
+	. "app.yhyue.com/moapp/jybase/mongodb"
+)
+
+var (
+	Bxt_raw_user *bxt_raw_user
+)
+
+type bxt_raw_user struct {
+	User_id        string //userid
+	Reg_time       string //注册日期
+	Reg_type       string //注册方式[手机、微信]
+	Province       string //省份
+	City           string //城市
+	Device         string //常用设备
+	Company        string //公司名称
+	Job            string //职务
+	Source_module  string //来源模块
+	Source_channel string //来源渠道
+	Follow_status  int    //关注状态  1未取关 2取关
+	Phone          string //手机号
+	Open_id        string //微信id
+	Channel_id     string //例如：如果是推荐人记推荐用户id
+	Email          string //
+	Name           string //姓名
+	Timestamp      string //更新时间
+}
+
+func (b *bxt_raw_user) TableName() string {
+	return "raw_user"
+}
+
+//
+func (b *bxt_raw_user) SaveFields() []string {
+	return []string{"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) selectField() map[string]interface{} {
+	return map[string]interface{}{
+		"_id":                 1,
+		"s_phone":             1,
+		"l_registedate":       1,
+		"s_company":           1,
+		"o_member_jy.s_email": 1,
+	}
+}
+
+//
+func (b *bxt_raw_user) Run(start_unix, end_unix int64, start_layout, end_layout string) {
+	log.Println("开始同步标讯通", b.TableName(), "表。。。", start_unix, end_unix)
+	b.add()
+	if start_unix > 0 {
+		b.update(start_unix, end_unix)
+	}
+	log.Println("同步讯通", b.TableName(), "表结束。。。")
+}
+
+//新增
+func (b *bxt_raw_user) add() (id string) {
+	defer Catch()
+	index := 0
+	array := []interface{}{}
+	lock := &sync.Mutex{}
+	pool := make(chan bool, Config.SelectMgoUserPool)
+	wait := &sync.WaitGroup{}
+	sess := Mgo.GetMgoConn()
+	defer Mgo.DestoryMongoConn(sess)
+	q := map[string]interface{}{
+		"i_appid": 2,
+	}
+	if TimeTask.Bxt_user_mgo_mysql_id != "" {
+		q["_id"] = map[string]interface{}{
+			"$gt": StringTOBsonId(TimeTask.Bxt_user_mgo_mysql_id),
+		}
+	}
+	log.Println("开始同步新增", b.TableName(), "表。。。", q)
+	it := sess.DB("bxt").C("user").Find(q).Sort("_id").Select(b.selectField()).Iter()
+	for temp := make(map[string]interface{}); it.Next(&temp); {
+		TimeTask.Bxt_user_mgo_mysql_id = BsonIdToSId(temp["_id"])
+		pool <- true
+		wait.Add(1)
+		go func(m map[string]interface{}) {
+			defer Catch()
+			defer func() {
+				<-pool
+				wait.Done()
+			}()
+			ru := new_bxt_raw_user(m, true)
+			lock.Lock()
+			defer lock.Unlock()
+			index++
+			array = append(array, ru.User_id, ru.Reg_time, ru.Province, ru.City, ru.Reg_type, ru.Device, ru.Company, ru.Job, ru.Source_module, ru.Source_channel, ru.Follow_status, ru.Phone, ru.Open_id, ru.Channel_id, ru.Name, ru.Email, ru.Timestamp)
+			if index%Config.InsertBathSize == 0 {
+				log.Println("同步新增", b.TableName(), "表", index)
+				Mysql_Main_Bxt.InsertIgnoreBatch(b.TableName(), b.SaveFields(), array)
+				array = []interface{}{}
+			}
+		}(temp)
+		temp = make(map[string]interface{})
+	}
+	wait.Wait()
+	if len(array) > 0 {
+		Mysql_Main_Bxt.InsertIgnoreBatch(b.TableName(), b.SaveFields(), array)
+		array = []interface{}{}
+	}
+	log.Println("同步新增", b.TableName(), "表结束。。。", index)
+	return
+}
+
+//更新
+func (b *bxt_raw_user) update(start_unix, end_unix int64) {
+	log.Println("开始同步user表，Mgo to Mysql 更新。。。")
+	defer Catch()
+	index := 0
+	fields := []string{"user_id", "province", "city", "reg_type", "device", "company", "job", "source_module", "source_channel", "follow_status", "phone", "openid", "name", "email", "timestamp"}
+	array := [][]interface{}{}
+	lock := &sync.Mutex{}
+	pool := make(chan bool, Config.SelectMgoUserPool)
+	wait := &sync.WaitGroup{}
+	sess := Mgo.GetMgoConn()
+	defer Mgo.DestoryMongoConn(sess)
+	q := map[string]interface{}{
+		"i_appid": 2,
+		"auto_updatetime": map[string]interface{}{
+			"$gte": start_unix,
+			"$lt":  end_unix,
+		},
+	}
+	log.Println("同步更新", b.TableName(), "表。。。", q)
+	it := sess.DB("bxt").C("user").Find(q).Sort("_id").Select(b.selectField()).Iter()
+	for temp := make(map[string]interface{}); it.Next(&temp); {
+		pool <- true
+		wait.Add(1)
+		go func(m map[string]interface{}) {
+			defer Catch()
+			defer func() {
+				<-pool
+				wait.Done()
+			}()
+			ru := new_bxt_raw_user(m, false)
+			lock.Lock()
+			defer lock.Unlock()
+			index++
+			array = append(array, []interface{}{ru.User_id, ru.Province, ru.City, ru.Reg_type, ru.Device, ru.Company, ru.Job, ru.Source_module, ru.Source_channel, ru.Follow_status, ru.Phone, ru.Open_id, ru.Name, ru.Email, ru.Timestamp})
+			if index%Config.UpdateBathSize == 0 {
+				log.Println("同步更新", b.TableName(), "表", index)
+				Mysql_Main_Bxt.UpdateBath(b.TableName(), fields, array)
+				array = [][]interface{}{}
+			}
+		}(temp)
+		temp = make(map[string]interface{})
+	}
+	wait.Wait()
+	if len(array) > 0 {
+		Mysql_Main_Bxt.UpdateBath(b.TableName(), fields, array)
+		array = [][]interface{}{}
+	}
+	log.Println("同步更新", b.TableName(), "表结束。。。", index)
+}
+
+//
+func new_bxt_raw_user(m map[string]interface{}, flag bool) *bxt_raw_user {
+	registedate := Int64All(m["l_registedate"])
+	//邮箱
+	email := strings.TrimSpace(ObjToString(m["s_myemail"]))
+	if email == "" {
+		o_member_jy, _ := m["o_member_jy"].(map[string]interface{})
+		email = strings.TrimSpace(ObjToString(o_member_jy["s_email"]))
+	}
+	return &bxt_raw_user{
+		User_id:       BsonIdToSId(m["_id"]),
+		Reg_time:      FormatDateByInt64(&registedate, Date_Full_Layout),
+		Reg_type:      "手机",
+		Company:       strings.TrimSpace(ObjToString(m["s_company"])),
+		Source_module: ObjToString(m["s_module"]),
+		Phone:         strings.TrimSpace(ObjToString(m["s_phone"])),
+		Email:         email,
+		Timestamp:     NowFormat(Date_Full_Layout),
+		Name:          strings.TrimSpace(ObjToString(m["s_name"])),
+	}
+}

online_datasync/main.go

@@ -17,14 +17,15 @@ var (
 	wait   = &sync.WaitGroup{}
 	pool   = make(chan bool, Config.SyncPool)
 	allMap = map[string]Entity{
-		"raw_user":    Raw_user,
-		"raw_product": Raw_product,
+		"raw_user":     Raw_user,
+		"bxt_raw_user": Bxt_raw_user,
+		//"raw_product": Raw_product,
 		//"action_order":          Action_order,
 		//"action_order_spec":     Action_order,
 		//"action_product_record": Action_product_record,
-		"user_prize":    User_prize,
+		//"user_prize":    User_prize,
 		"integral_flow": Integral_flow,
-		"point_type":    Point_type,
+		//"point_type":    Point_type,
 	}
 )
 

online_syncto_offline/config.json

@@ -3,5 +3,6 @@
 	"updateBathSize": 100,
 	"insertBathSize": 200,
 	"selectBathSize": 200,
-	"syncPool":5
+	"syncPool":5,
+	"duration": 120
 }

online_syncto_offline/config/config.go

@@ -8,6 +8,7 @@ type config struct {
 	InsertBathSize int
 	SelectBathSize int
 	SyncPool       int
+	Duration       int
 }
 type timeTask struct {
 	Datetime                 string `json:"datetime"`
@@ -22,6 +23,7 @@ type timeTask struct {
 	Raw_page_id              int64  `json:"raw_page_id"`
 	Raw_product_id           int64  `json:"raw_product_id"`
 	Raw_user_id              int64  `json:"raw_user_id"`
+	Bxt_raw_user_id          int64  `json:"bxt_raw_user_id"`
 	Point_type_id            int64  `json:"point_type_id"`
 	Integral_flow_id         int64  `json:"integral_flow_id"`
 	User_prize_id            int64  `json:"user_prize_id"`

online_syncto_offline/db.json

@@ -8,7 +8,7 @@
 		}
 	},
     "mysql": {
-	    "main": {
+	    "main_jianyu": {
 	        "dbName": "jianyu",
 	        "address": "192.168.3.166:3306",
 	        "userName": "root",
@@ -16,13 +16,29 @@
 			"maxOpenConns": 5,
 			"maxIdleConns": 5
 	    },
-		"to": {
+		"main_bxt": {
+	        "dbName": "bxt",
+	        "address": "192.168.3.166:3306",
+	        "userName": "root",
+	        "passWord": "Topnet@123",
+			"maxOpenConns": 5,
+			"maxIdleConns": 5
+	    },
+		"to_jianyu": {
 	        "dbName": "jianyu",
 	        "address": "192.168.3.71: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": "jianyu",
@@ -31,6 +47,30 @@
 	        "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
 	    }
     }
 }

online_syncto_offline/db/db.go

@@ -9,11 +9,16 @@ import (
 )
 
 var (
-	DbConf            *dbConf
-	Mysql_Main        *mysql.Mysql
-	Mysql_To          *mysql.Mysql
-	Mysql_From_Jianyu *mysql.Mysql
-	Mgo               MongodbSim
+	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
 )
 
 type dbConf struct {
@@ -21,9 +26,14 @@ type dbConf struct {
 		Main *mgoConf
 	}
 	Mysql struct {
-		Main        *mysqlConf
-		To          *mysqlConf
-		From_jianyu *mysqlConf
+		Main_jianyu       *mysqlConf
+		Main_bxt          *mysqlConf
+		To_jianyu         *mysqlConf
+		From_jianyu       *mysqlConf
+		To_bxt            *mysqlConf
+		From_bxt          *mysqlConf
+		From_jyactivities *mysqlConf
+		From_jypoints     *mysqlConf
 	}
 }
 type mgoConf struct {
@@ -57,29 +67,53 @@ func init() {
 			}
 			Mgo.InitPool()
 		}
-		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,
+		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.Init()
+			Mysql_Main_Jianyu.Init()
 		}
-		if DbConf.Mysql.To != nil {
-			log.Println("初始化 mysql to")
-			Mysql_To = &mysql.Mysql{
-				Address:      DbConf.Mysql.To.Address,
-				UserName:     DbConf.Mysql.To.UserName,
-				PassWord:     DbConf.Mysql.To.PassWord,
-				DBName:       DbConf.Mysql.To.DbName,
-				MaxOpenConns: DbConf.Mysql.To.MaxOpenConns,
-				MaxIdleConns: DbConf.Mysql.To.MaxIdleConns,
+		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_To.Init()
+			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")
@@ -93,5 +127,41 @@ func init() {
 			}
 			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()
+		}
 	}
 }

online_syncto_offline/entity/action_event.go

@@ -23,5 +23,5 @@ func (a *action_event) SaveFields() []string {
 
 //
 func (a *action_event) Run(start_layout, end_layout string) {
-	TimeTask.Action_event_id = sync_add(Mysql_Main, a.TableName(), a.SaveFields(), TimeTask.Action_event_id, nil)
+	TimeTask.Action_event_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), a.SaveFields(), TimeTask.Action_event_id, nil)
 }

online_syncto_offline/entity/action_login.go

@@ -23,5 +23,5 @@ func (a *action_login) SaveFields() []string {
 
 //
 func (a *action_login) Run(start_layout, end_layout string) {
-	TimeTask.Action_login_id = sync_add(Mysql_Main, a.TableName(), a.SaveFields(), TimeTask.Action_login_id, nil)
+	TimeTask.Action_login_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), a.SaveFields(), TimeTask.Action_login_id, nil)
 }

online_syncto_offline/entity/action_order.go

@@ -24,7 +24,7 @@ func (a *action_order) SaveFields() []string {
 //
 func (a *action_order) Run(start_layout, end_layout string) {
 	if start_layout != "" {
-		sync_update(a.TableName(), start_layout, end_layout, a.SaveFields(), TimeTask.Action_order_id)
+		sync_update(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), start_layout, end_layout, a.SaveFields(), TimeTask.Action_order_id)
 	}
-	TimeTask.Action_order_id = sync_add(Mysql_Main, a.TableName(), a.SaveFields(), TimeTask.Action_order_id, nil)
+	TimeTask.Action_order_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), a.SaveFields(), TimeTask.Action_order_id, nil)
 }

online_syncto_offline/entity/action_order_spec.go

@@ -23,5 +23,5 @@ func (a *action_order_spec) SaveFields() []string {
 
 //
 func (a *action_order_spec) Run(start_layout, end_layout string) {
-	TimeTask.Action_order_spec_id = sync_add(Mysql_Main, a.TableName(), a.SaveFields(), TimeTask.Action_order_spec_id, nil)
+	TimeTask.Action_order_spec_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), a.SaveFields(), TimeTask.Action_order_spec_id, nil)
 }

online_syncto_offline/entity/action_page.go

@@ -28,7 +28,7 @@ func (a *action_page) SaveFields() []string {
 //
 func (a *action_page) Run(start_layout, end_layout string) {
 	if start_layout != "" {
-		sync_update(a.TableName(), start_layout, end_layout, a.SaveFields(), TimeTask.Action_page_id)
+		sync_update(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), start_layout, end_layout, a.SaveFields(), TimeTask.Action_page_id)
 	}
 	log.Println("开始加载超级订阅、大会员状态。。。")
 	sess := Mgo.GetMgoConn()
@@ -68,7 +68,7 @@ func (a *action_page) Run(start_layout, end_layout string) {
 		m = make(map[string]interface{})
 	}
 	log.Println("加载超级订阅、大会员状态结束。。。", index)
-	TimeTask.Action_page_id = sync_add(Mysql_Main, a.TableName(), a.SaveFields(), TimeTask.Action_page_id, func(fv map[string]interface{}) {
+	TimeTask.Action_page_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), a.SaveFields(), TimeTask.Action_page_id, func(fv map[string]interface{}) {
 		userId, _ := fv["user_id"].(string)
 		if userId != "" && all[userId] != nil {
 			fv["user_status_big"] = all[userId].memberStatus

online_syncto_offline/entity/action_product_record.go

@@ -24,7 +24,7 @@ func (a *action_product_record) SaveFields() []string {
 //
 func (a *action_product_record) Run(start_layout, end_layout string) {
 	if start_layout != "" {
-		sync_update(a.TableName(), start_layout, end_layout, a.SaveFields(), TimeTask.Action_product_record_id)
+		sync_update(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), start_layout, end_layout, a.SaveFields(), TimeTask.Action_product_record_id)
 	}
-	TimeTask.Action_product_record_id = sync_add(Mysql_Main, a.TableName(), a.SaveFields(), TimeTask.Action_product_record_id, nil)
+	TimeTask.Action_product_record_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, a.TableName(), a.SaveFields(), TimeTask.Action_product_record_id, nil)
 }

online_syncto_offline/entity/bxt_dataexport_order.go

@@ -0,0 +1,26 @@
+package entity
+
+import (
+	. "online_syncto_offline/db"
+)
+
+var (
+	Bxt_dataexport_order *bxt_dataexport_order
+)
+
+type bxt_dataexport_order struct {
+}
+
+func (b *bxt_dataexport_order) TableName() string {
+	return "dataexport_order"
+}
+
+//
+func (b *bxt_dataexport_order) SaveFields() []string {
+	return nil
+}
+
+//
+func (b *bxt_dataexport_order) Run(start_layout, end_layout string) {
+	sync_full(b, Mysql_From_Bxt, Mysql_To_Bxt)
+}

online_syncto_offline/entity/bxt_raw_user.go

@@ -0,0 +1,30 @@
+package entity
+
+import (
+	. "online_syncto_offline/config"
+	. "online_syncto_offline/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)
+	}
+	TimeTask.Bxt_raw_user_id = sync_add(Mysql_Main_Bxt, Mysql_To_Bxt, b.TableName(), b.SaveFields(), TimeTask.Bxt_raw_user_id, nil)
+}

online_syncto_offline/entity/contract.go

@@ -0,0 +1,26 @@
+package entity
+
+import (
+	. "online_syncto_offline/db"
+)
+
+var (
+	Contract *contract
+)
+
+type contract struct {
+}
+
+func (c *contract) TableName() string {
+	return "contract"
+}
+
+//
+func (c *contract) SaveFields() []string {
+	return nil
+}
+
+//
+func (c *contract) Run(start_layout, end_layout string) {
+	sync_full(c, Mysql_From_Jianyu, Mysql_To_Jianyu)
+}

online_syncto_offline/entity/dataexport_order.go

@@ -1,5 +1,9 @@
 package entity
 
+import (
+	. "online_syncto_offline/db"
+)
+
 var (
 	Dataexport_order *dataexport_order
 )
@@ -18,5 +22,5 @@ func (d *dataexport_order) SaveFields() []string {
 
 //
 func (d *dataexport_order) Run(start_layout, end_layout string) {
-	sync_full(d)
+	sync_full(d, Mysql_From_Jianyu, Mysql_To_Jianyu)
 }

online_syncto_offline/entity/entity.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"log"
 	. "online_syncto_offline/config"
-	. "online_syncto_offline/db"
 	"time"
 
 	. "app.yhyue.com/moapp/jybase/common"
@@ -18,7 +17,7 @@ type Entity interface {
 }
 
 //
-func sync_add(m *mysql.Mysql, tableName string, saveField []string, id int64, f func(fv map[string]interface{})) (lastId int64) {
+func sync_add(main, to *mysql.Mysql, tableName string, saveField []string, id int64, f func(fv map[string]interface{})) (lastId int64) {
 	log.Println("开始同步", tableName, "表。。。")
 	index := 0
 	array := []interface{}{}
@@ -29,7 +28,7 @@ func sync_add(m *mysql.Mysql, tableName string, saveField []string, id int64, f
 	} else {
 		q = fmt.Sprintf(`select * from %s where id>%d order by id`, tableName, id)
 	}
-	m.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
+	main.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
 		for _, v := range *l {
 			index++
 			lastId = Int64All(v["id"])
@@ -41,13 +40,13 @@ func sync_add(m *mysql.Mysql, tableName string, saveField []string, id int64, f
 			}
 			if index%Config.InsertBathSize == 0 {
 				log.Println("同步", tableName, "表", index)
-				Mysql_To.InsertIgnoreBatch(tableName, saveField, array)
+				to.InsertIgnoreBatch(tableName, saveField, array)
 				array = []interface{}{}
 			}
 		}
 	}, q)
 	if len(array) > 0 {
-		Mysql_To.InsertIgnoreBatch(tableName, saveField, array)
+		to.InsertIgnoreBatch(tableName, saveField, array)
 		array = []interface{}{}
 	}
 	log.Println("同步", tableName, "表结束。。。", index)
@@ -55,11 +54,11 @@ func sync_add(m *mysql.Mysql, tableName string, saveField []string, id int64, f
 }
 
 //
-func sync_update(tableName, start_layout, end_layout string, saveFields []string, id int64) {
+func sync_update(main, to *mysql.Mysql, tableName, start_layout, end_layout string, saveFields []string, id int64) {
 	log.Println("开始同步更新", tableName, "表 。。。")
 	index := 0
 	array := [][]interface{}{}
-	Mysql_Main.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
+	main.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
 		for _, v := range *l {
 			index++
 			datas := []interface{}{}
@@ -69,35 +68,35 @@ func sync_update(tableName, start_layout, end_layout string, saveFields []string
 			array = append(array, datas)
 			if index%Config.UpdateBathSize == 0 {
 				log.Println("同步更新", tableName, "表", index)
-				Mysql_To.UpdateBath(tableName, saveFields, array)
+				to.UpdateBath(tableName, saveFields, array)
 				array = [][]interface{}{}
 			}
 		}
 	}, fmt.Sprintf(`select * from %s where id<? and timestamp>=? and timestamp<?`, tableName), id, start_layout, end_layout)
 	if len(array) > 0 {
-		Mysql_To.UpdateBath(tableName, saveFields, array)
+		to.UpdateBath(tableName, saveFields, array)
 		array = [][]interface{}{}
 	}
 	log.Println("同步更新", tableName, "表结束。。。", index)
 }
 
 //全量同步
-func sync_full(e Entity) {
-	if Mysql_To.Address != "172.17.4.195:3366" || Mysql_To.DBName != "buryingpoint" {
+func sync_full(e Entity, from, to *mysql.Mysql) {
+	if to.Address != "172.17.4.195:3366" || (to.DBName != "buryingpoint" && to.DBName != "bxt") {
 		log.Println("数据库配置错误")
 		return
 	}
-	_, err := Mysql_To.ExecBySql(`drop table ` + e.TableName())
+	_, err := to.ExecBySql(`drop table ` + e.TableName())
 	if err == nil {
 		log.Println(e.TableName(), "清空表完成")
 	} else {
 		log.Println(e.TableName(), "清空表出错", err)
 	}
-	list := Mysql_From_Jianyu.SelectBySql(`SHOW CREATE TABLE ` + e.TableName())
+	list := from.SelectBySql(`SHOW CREATE TABLE ` + e.TableName())
 	if list != nil && len(*list) == 1 {
 		createTable := (*list)[0]["Create Table"].(string)
 		log.Println(e.TableName(), "建表语句", createTable)
-		_, err := Mysql_To.ExecBySql(createTable)
+		_, err := to.ExecBySql(createTable)
 		if err == nil {
 			log.Println(e.TableName(), "表创建成功")
 		} else {
@@ -108,7 +107,7 @@ func sync_full(e Entity) {
 		return
 	}
 	var lastId int64
-	array := Mysql_From_Jianyu.SelectBySql(`select id from ` + e.TableName() + ` order by id desc limit 1`)
+	array := from.SelectBySql(`select id from ` + e.TableName() + ` order by id desc limit 1`)
 	if len(*array) == 1 {
 		lastId = (*array)[0]["id"].(int64)
 	} else {
@@ -116,12 +115,12 @@ func sync_full(e Entity) {
 		return
 	}
 	fields := []string{}
-	fields_list := Mysql_From_Jianyu.SelectBySql(`describe ` + e.TableName())
+	fields_list := from.SelectBySql(`describe ` + e.TableName())
 	for _, v := range *fields_list {
 		fields = append(fields, v["Field"].(string))
 	}
 	for {
-		id := sync_add(Mysql_From_Jianyu, e.TableName(), fields, -1, nil)
+		id := sync_add(from, to, e.TableName(), fields, -1, nil)
 		if id >= lastId {
 			break
 		}

online_syncto_offline/entity/integral_flow.go

@@ -23,5 +23,5 @@ func (i *integral_flow) SaveFields() []string {
 
 //
 func (i *integral_flow) Run(start_layout, end_layout string) {
-	TimeTask.Integral_flow_id = sync_add(Mysql_Main, i.TableName(), i.SaveFields(), TimeTask.Integral_flow_id, nil)
+	TimeTask.Integral_flow_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, i.TableName(), i.SaveFields(), TimeTask.Integral_flow_id, nil)
 }

online_syncto_offline/entity/point_type.go

@@ -1,7 +1,6 @@
 package entity
 
 import (
-	. "online_syncto_offline/config"
 	. "online_syncto_offline/db"
 )
 
@@ -18,10 +17,10 @@ func (p *point_type) TableName() string {
 
 //
 func (p *point_type) SaveFields() []string {
-	return []string{"id", "code", "name"}
+	return nil
 }
 
 //
 func (p *point_type) Run(start_layout, end_layout string) {
-	TimeTask.Point_type_id = sync_add(Mysql_Main, p.TableName(), p.SaveFields(), TimeTask.Point_type_id, nil)
+	sync_full(p, Mysql_From_Jypoints, Mysql_To_Jianyu)
 }

online_syncto_offline/entity/raw_element.go

@@ -23,5 +23,5 @@ func (r *raw_element) SaveFields() []string {
 
 //
 func (r *raw_element) Run(start_layout, end_layout string) {
-	TimeTask.Raw_element_id = sync_add(Mysql_Main, r.TableName(), r.SaveFields(), TimeTask.Raw_element_id, nil)
+	TimeTask.Raw_element_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, r.TableName(), r.SaveFields(), TimeTask.Raw_element_id, nil)
 }

online_syncto_offline/entity/raw_event.go

@@ -23,5 +23,5 @@ func (r *raw_event) SaveFields() []string {
 
 //
 func (r *raw_event) Run(start_layout, end_layout string) {
-	TimeTask.Raw_event_id = sync_add(Mysql_Main, r.TableName(), r.SaveFields(), TimeTask.Raw_event_id, nil)
+	TimeTask.Raw_event_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, r.TableName(), r.SaveFields(), TimeTask.Raw_event_id, nil)
 }

online_syncto_offline/entity/raw_page.go

@@ -23,5 +23,5 @@ func (r *raw_page) SaveFields() []string {
 
 //
 func (r *raw_page) Run(start_layout, end_layout string) {
-	TimeTask.Raw_page_id = sync_add(Mysql_Main, r.TableName(), r.SaveFields(), TimeTask.Raw_page_id, nil)
+	TimeTask.Raw_page_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, r.TableName(), r.SaveFields(), TimeTask.Raw_page_id, nil)
 }

online_syncto_offline/entity/raw_product.go

@@ -23,5 +23,5 @@ func (r *raw_product) SaveFields() []string {
 
 //
 func (r *raw_product) Run(start_layout, end_layout string) {
-	TimeTask.Action_product_record_id = sync_add(Mysql_Main, r.TableName(), r.SaveFields(), TimeTask.Action_product_record_id, nil)
+	TimeTask.Action_product_record_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, r.TableName(), r.SaveFields(), TimeTask.Action_product_record_id, nil)
 }

online_syncto_offline/entity/raw_user.go

@@ -24,7 +24,7 @@ func (r *raw_user) SaveFields() []string {
 //
 func (r *raw_user) Run(start_layout, end_layout string) {
 	if start_layout != "" {
-		sync_update(r.TableName(), start_layout, end_layout, r.SaveFields(), TimeTask.Raw_user_id)
+		sync_update(Mysql_Main_Jianyu, Mysql_To_Jianyu, r.TableName(), start_layout, end_layout, r.SaveFields(), TimeTask.Raw_user_id)
 	}
-	TimeTask.Raw_user_id = sync_add(Mysql_Main, r.TableName(), r.SaveFields(), TimeTask.Raw_user_id, nil)
+	TimeTask.Raw_user_id = sync_add(Mysql_Main_Jianyu, Mysql_To_Jianyu, r.TableName(), r.SaveFields(), TimeTask.Raw_user_id, nil)
 }

online_syncto_offline/entity/refund_record.go

@@ -1,5 +1,9 @@
 package entity
 
+import (
+	. "online_syncto_offline/db"
+)
+
 var (
 	Refund_record *refund_record
 )
@@ -18,5 +22,5 @@ func (r *refund_record) SaveFields() []string {
 
 //
 func (r *refund_record) Run(start_layout, end_layout string) {
-	sync_full(r)
+	sync_full(r, Mysql_From_Jianyu, Mysql_To_Jianyu)
 }

online_syncto_offline/entity/return_money_record.go

@@ -1,5 +1,9 @@
 package entity
 
+import (
+	. "online_syncto_offline/db"
+)
+
 var (
 	Return_money_record *return_money_record
 )
@@ -18,5 +22,5 @@ func (r *return_money_record) SaveFields() []string {
 
 //
 func (r *return_money_record) Run(start_layout, end_layout string) {
-	sync_full(r)
+	sync_full(r, Mysql_From_Jianyu, Mysql_To_Jianyu)
 }

online_syncto_offline/entity/user_prize.go

@@ -1,7 +1,6 @@
 package entity
 
 import (
-	. "online_syncto_offline/config"
 	. "online_syncto_offline/db"
 )
 
@@ -18,13 +17,10 @@ func (u *user_prize) TableName() string {
 
 //
 func (u *user_prize) SaveFields() []string {
-	return []string{"id", "userId", "prizeId", "validityDates", "beginDate", "endDate", "useDate", "prizeType", "appId", "lotteryId", "createTime", "name", "full", "reduce", "userName", "orderCode", "timestamp"}
+	return nil
 }
 
 //
 func (u *user_prize) Run(start_layout, end_layout string) {
-	if start_layout != "" {
-		sync_update(u.TableName(), start_layout, end_layout, u.SaveFields(), TimeTask.User_prize_id)
-	}
-	TimeTask.User_prize_id = sync_add(Mysql_Main, u.TableName(), u.SaveFields(), TimeTask.User_prize_id, nil)
+	sync_full(u, Mysql_From_Jyactivities, Mysql_To_Jianyu)
 }

online_syncto_offline/main.go

@@ -7,33 +7,42 @@ import (
 	. "online_syncto_offline/config"
 	. "online_syncto_offline/entity"
 	"sync"
+	"time"
 
 	. "app.yhyue.com/moapp/jybase/common"
 	. "app.yhyue.com/moapp/jybase/date"
 )
 
 var (
-	wait   = &sync.WaitGroup{}
-	pool   = make(chan bool, Config.SyncPool)
-	allMap = map[string]Entity{
-		"raw_user":     Raw_user,
-		"raw_product":  Raw_product,
-		"raw_page":     Raw_page,
-		"raw_element":  Raw_element,
-		"raw_event":    Raw_event,
+	wait       = &sync.WaitGroup{}
+	pool       = make(chan bool, Config.SyncPool)
+	incSyncMap = map[string]Entity{
+		//jianyu
+		"raw_user": Raw_user,
+		//"raw_product":  Raw_product,
+		//"raw_page":     Raw_page,
+		//"raw_element":  Raw_element,
+		//"raw_event":    Raw_event,
 		"action_login": Action_login,
-		"action_page":  Action_page,
-		"action_event": Action_event,
+		//"action_page":  Action_page,
+		//"action_event": Action_event,
 		//"action_order":          Action_order,
 		//"action_order_spec":     Action_order_spec,
 		//"action_product_record": Action_product_record,
-		"user_prize":          User_prize,
-		"integral_flow":       Integral_flow,
-		"point_type":          Point_type,
+		"integral_flow": Integral_flow,
+		//bxt
+		"bxt_raw_user": Bxt_raw_user,
+	}
+	fullSyncMap = map[string]Entity{
+		//jianyu
 		"dataexport_order":    Dataexport_order,
 		"refund_record":       Refund_record,
 		"return_money_record": Return_money_record,
 		"contract":            Contract,
+		"point_type":          Point_type,
+		"user_prize":          User_prize,
+		//bxt
+		"bxt_dataexport_order": Bxt_dataexport_order,
 	}
 )
 
@@ -41,31 +50,63 @@ func main() {
 	m := flag.Int("m", 0, "模式 0：定时任务 1：非定时任务")
 	t := flag.String("t", "", "表名")
 	flag.Parse()
-	all := []Entity{}
-	for _, v := range allMap {
-		all = append(all, v)
+	inc := []Entity{}
+	for _, v := range incSyncMap {
+		inc = append(inc, v)
+	}
+	full := []Entity{}
+	for _, v := range fullSyncMap {
+		full = append(full, v)
 	}
 	if *m == 1 {
 		if *t == "" {
-			run(all...)
-		} else if allMap[*t] != nil {
-			run(allMap[*t])
+			incSyncRun(inc...)
+			fullSyncRun(full...)
+		} else if incSyncMap[*t] != nil {
+			incSyncRun(incSyncMap[*t])
+		} else if fullSyncMap[*t] != nil {
+			fullSyncRun(fullSyncMap[*t])
 		} else {
 			log.Fatalln(*t, "表的同步数据任务不存在")
 		}
 	} else {
-		SimpleCrontab(false, Config.RunTime, func() {
-			run(all...)
+		go func() {
+			now := time.Now()
+			next := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local)
+			next = next.Add(time.Minute * time.Duration(Config.Duration))
+			timer := time.NewTimer(next.Sub(now))
+			for {
+				select {
+				case <-timer.C:
+					timer.Reset(time.Minute * time.Duration(Config.Duration))
+					fullSyncRun(full...)
+				}
+			}
+		}()
+		go SimpleCrontab(false, Config.RunTime, func() {
+			incSyncRun(inc...)
 		})
 		select {}
 	}
 }
 
+func fullSyncRun(es ...Entity) {
+	log.Println("全量同步任务开始。。。", fmt.Sprintf("%+v", TimeTask))
+	run("", "", es...)
+	log.Println("全量同步任务结束。。。")
+}
+
 //
-func run(es ...Entity) {
+func incSyncRun(es ...Entity) {
 	end_layout := NowFormat(Date_Full_Layout)
-	log.Println("同步任务开始。。。", fmt.Sprintf("%+v", TimeTask))
-	//
+	log.Println("增量同步任务开始。。。", fmt.Sprintf("%+v", TimeTask))
+	run(TimeTask.Datetime, end_layout, es...)
+	TimeTask.Datetime = end_layout
+	WriteSysConfig("./timetask.json", &TimeTask)
+	log.Println("增量同步任务结束。。。")
+}
+
+func run(d1, d2 string, es ...Entity) {
 	for _, e := range es {
 		pool <- true
 		wait.Add(1)
@@ -74,12 +115,8 @@ func run(es ...Entity) {
 				<-pool
 				wait.Done()
 			}()
-			v.Run(TimeTask.Datetime, end_layout)
+			v.Run(d1, d2)
 		}(e)
 	}
-	//
 	wait.Wait()
-	TimeTask.Datetime = end_layout
-	WriteSysConfig("./timetask.json", &TimeTask)
-	log.Println("同步任务结束。。。")
 }

online_syncto_offline/timetask.json

@@ -1 +1 @@
-{"datetime":"2021-05-06 11:55:12","action_event_id":213662781,"action_login_id":75312,"action_order_id":517131,"action_order_spec_id":157183,"action_page_id":213662872,"action_product_record_id":157183,"raw_element_id":1851,"raw_event_id":5,"raw_page_id":327,"raw_product_id":0,"raw_user_id":1933484,"point_type_id":12,"integral_flow_id":2601,"user_prize_id":699,"dataexport_order_id":0,"refund_record_id":0,"return_money_record_id":0}
+{"datetime":"2021-05-06 11:55:12","action_login_id":75312,"raw_user_id":1933484,"bxt_raw_user_id":1933484,"integral_flow_id":2601}