buryingPoint/data_analysis/main.go

package main

import (
	. "data_analysis/config"
	. "data_analysis/db"
	. "data_analysis/entity"
	"flag"
	"fmt"
	"log"
	"sync"
	"time"

	. "app.yhyue.com/moapp/jybase/common"
	. "app.yhyue.com/moapp/jybase/date"
)

var (
	incSyncMap = map[string]Entity{
		//jianyu
		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{}
	incNewSyncMap = map[string]Entity{}
)

func init() {
	for _, v := range Config.TableName {
		fullSyncMap[v] = NewFinalJob(v, "")
	}
	for _, v := range Config.IncTask.TableName {
		incNewSyncMap[v] = NewFinalJob(v, Config.IncTask.IncField)
	}
}

func main() {
	m := flag.Int("m", 0, "模式 0：定时任务 1：非定时任务")
	t := flag.String("t", "", "表名")
	flag.Parse()
	if *m == 1 {
		if *t == "" {
			incSyncRun(incSyncMap)
			fullSyncRun(fullSyncMap)
			incNewSyncRun(incNewSyncMap)
		} else if incSyncMap[*t] != nil {
			incSyncRun(map[string]Entity{
				*t: incSyncMap[*t],
			})
		} else if fullSyncMap[*t] != nil {
			fullSyncRun(map[string]Entity{
				*t: fullSyncMap[*t],
			})
		} else if incNewSyncMap[*t] != nil {
			incNewSyncRun(map[string]Entity{
				*t: incNewSyncMap[*t],
			})
		} else {
			log.Fatalln(*t, "表的同步数据任务不存在")
		}
	} else {
		start()
		startNew()
		select {}
	}
}

var isRuning bool

func start() {
	go func() {
		if isRuning {
			SendMail("同步数据任务上轮未结束，请排查！")
			return
		}
		isRuning = true
		if err := Mysql_Main.DB.Ping(); err != nil {
			log.Println("Mysql_Main Ping", err)
			Mysql_Main.Init()
		}
		incSyncRun(incSyncMap)
		fullSyncRun(fullSyncMap)
		isRuning = false
	}()
	now := time.Now()
	next := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local)
	next = next.Add(time.Minute * Config.Duration)
	time.AfterFunc(next.Sub(now), start)
}
func startNew() {
	if err := Mysql_Main.DB.Ping(); err != nil {
		log.Println("Mysql_Main Ping", err)
		Mysql_Main.Init()
	}
	incNewSyncRun(incNewSyncMap)
	time.AfterFunc(time.Minute*Config.IncTask.Duration, startNew)
}

func fullSyncRun(es map[string]Entity) {
	log.Println("全量同步任务开始。。。")
	run("", "", es)
	log.Println("全量同步任务结束。。。")
}

//
func incSyncRun(es map[string]Entity) {
	end_layout := NowFormat(Date_Full_Layout)
	log.Println("增量同步任务开始。。。", fmt.Sprintf("%+v", TimeTask))
	run(TimeTask.Datetime, end_layout, es)
	TimeTask.Datetime = end_layout
	WriteSysConfig("./timetask.json", &TimeTask)
	log.Println("增量同步任务结束。。。")
}

//
func incNewSyncRun(es map[string]Entity) {
	end_layout := NowFormat(Date_Full_Layout)
	log.Println("新的增量同步任务开始。。。", fmt.Sprintf("%+v", IncTask))
	run(IncTask.Datetime, end_layout, es)
	IncTask.Datetime = end_layout
	WriteSysConfig("./inctask.json", &IncTask)
	log.Println("新的增量同步任务结束。。。")
}
func run(d1, d2 string, es map[string]Entity) {
	wait := &sync.WaitGroup{}
	pool := make(chan bool, Config.SyncPool)
	for _, e := range es {
		pool <- true
		wait.Add(1)
		go func(v Entity) {
			defer func() {
				<-pool
				wait.Done()
			}()
			v.Run(d1, d2)
		}(e)
	}
	wait.Wait()
}