dxp-core/dxp-engine/service/cron/dataCache.go

package cron

import (
	"DataShare/dao"
	"DataShare/global"
	"DataShare/model"
	"DataShare/service"
	"fmt"
	"github.com/jasonlvhit/gocron"
	"github.com/sirupsen/logrus"
	"strconv"
	"sync"
)

type CacheTimer struct{
	sc *gocron.Scheduler
	q chan bool

	Id  []int
}

var once sync.Once
var singleInstance *CacheTimer

func GetInstance() *CacheTimer {
	if singleInstance == nil {
		once.Do(
			func() {
				singleInstance = &CacheTimer{}
			})
	}

	return singleInstance
}

func (this *CacheTimer)Start(){
	//每次开机运行都执行一下daytask
	this.q = make(chan bool)
	this.Id = make([]int,0)
	go this.jobs(this.q)
}

func (this *CacheTimer)Stop(){
	this.q <- true
}

func (this *CacheTimer)jobs(quit <-chan bool) {
	for {
		//cron jobs
		this.sc = gocron.NewScheduler()
		//每天凌晨00:00 同步配置文件
		//this.sc.Every(1).Day().At("00:00:00").Do(this.CacheTask)
		fmt.Println("---------------------------timer jobs------------------------")
		this.sc.Every(5).Seconds().Do(this.CacheTask)

		this.sc.Every(10).Seconds().Do(this.SubscribeTask)

		select {
		case <-quit:
			// here we receive from quit and exit
			// if `g` has started, we may want to `g.Clear()`
			// or the scheduled jobs will continue, we will just not be waiting for them to finish.
			return
		case <-this.sc.Start():
			// here we know all the jobs are done, and go round the loop again
		}
	}
}

func (this *CacheTimer)CacheTask(){

	//this.sc.Clear()
	data_cache_list := dao.LoadDataCacheList()
	if len(this.Id) != len(data_cache_list) {
		this.sc.Clear()
		this.Id = make([]int,0)
		this.sc.Every(5).Seconds().Do(this.CacheTask)
	}

	for i:=0;i<len(data_cache_list);i++{

		exist := false
		for m:=0;m<len(this.Id);m++ {
			if this.Id[m] == data_cache_list[i].Id {
				exist = true
				break
			}
		}
		if exist == true {
			continue
		}

		global.SystemLogger.Log(logrus.InfoLevel,"-------------add new cache task-----------------")
		global.SystemLogger.Log(logrus.InfoLevel,data_cache_list[i])
		//fmt.Println(data_cache_list[i])

		cacheCycle := data_cache_list[i].CacheCycle
		//"unit": "d", // 周期单位，m分钟，h小时，d天，w周，m月，y年
		//"value": 1, // 周期值，每间隔多少周期单位执行一次
		val,err:= strconv.Atoi(cacheCycle.Value)
		if err != nil {
			global.SystemLogger.Log(logrus.ErrorLevel,"add cache task failed,err:"+err.Error())
			continue

		}
		this.Id = append(this.Id,data_cache_list[i].Id)
		fmt.Println("-----ids--------",this.Id)
		fmt.Println("----i-----------",i)
		//err = service.Business2_DataCache(&data_cache_list[i])

		var cacheTask model.DataCacheOfRedis
		cacheTask = data_cache_list[i]
		if cacheCycle.Unit == "d"{
			this.sc.Every(uint64(val)).Days().Do(func() {
				err = service.Business2_DataCache(&cacheTask)
			})
		}else if cacheCycle.Unit == "h"{
			this.sc.Every(uint64(val)).Hours().Do(func() {
				err = service.Business2_DataCache(&cacheTask)
			})
		}else if cacheCycle.Unit == "m"{
			this.sc.Every(uint64(val)).Minutes().Do(func() {
				err = service.Business2_DataCache(&cacheTask)
			})
		}else if cacheCycle.Unit == "w"{
			this.sc.Every(uint64(val)).Weeks().Do(func() {
				err = service.Business2_DataCache(&cacheTask)
			})
		}else if cacheCycle.Unit == "s"{
			this.sc.Every(uint64(val)).Seconds().Do(func() {
				err = service.Business2_DataCache(&cacheTask)
			})
		}

		if err != nil {
			global.SystemLogger.Log(logrus.ErrorLevel,fmt.Sprintf("serviceUrl:%s ,cache data failed,err msg:%s.",
				data_cache_list[i].ServiceConfig.ServiceUrl,err.Error()))

		}
	}

	//this.sc.Every(1).Day().At("00:00:00").Do(this.CacheTask)
}

func (this *CacheTimer)SubscribeTask(){
	CallSubscribe()
}