定时任务分master模式和非master模式

zhangqian

2023-10-17 f75b6f9ee6b2c542d52b4a11113bd281b421c4c0

定时任务分master模式和非master模式

 crontask/cron_task.go

@@ -16,7 +16,7 @@

var s *gocron.Scheduler

func StartTask() error {
func StartTask(isMaster bool) error {
    finishNumberTimeInterval := conf.Conf.PLC.FinishNumberTimeInterval
    totalNumberTimeInterval := conf.Conf.PLC.TotalNumberTimeInterval
    if finishNumberTimeInterval == 0 {
@@ -75,32 +75,21 @@

    })

    s.Every(60).Seconds().Do(SyncProductionProgress) //同步生产数据
    s.Every(30).Seconds().Do(SyncTaskStatus)         //同步任务状态
    if isMaster {
        s.Every(60).Seconds().Do(SyncProductionProgress) //同步生产数据
        s.Every(30).Seconds().Do(SyncTaskStatus)         //同步任务状态
    }
    s.StartAsync()
    return nil
}

func SyncProductionProgress() {
    plcConfig, code := service.NewDevicePlcService().GetDevicePlc()
    if code != ecode.OK {
    progressList, err := service.NewProgressService().GetProgressList()
    if err != nil {
        logx.Errorf("SyncProductionProgress get records err:%v", err)
        return
    }
    var channels []int32
    for _, item := range plcConfig.Details {
        if item.FieldName == constvar.PlcStartAddressTypeFinishNumber {
            channels = append(channels, item.Channel)
        }
    }
    for _, channel := range channels {
        progress, err := service.NewProgressService().GetCurrentProgress(channel)
        if err != nil {
            return
        }
        if progress == nil {
            return
        }

    for _, progress := range progressList {
        caller := nsq.NewCaller(fmt.Sprintf(constvar.NsqTopicSyncTaskProgress, conf.Conf.NsqConf.NodeId), "")
        err = caller.Send(progress)
        if err != nil {
@@ -130,14 +119,18 @@
    }
}

func StopTask() {
func RestartTask(isMaster bool) error {
    if s != nil {
        s.Stop()
    }
    return StartTask(isMaster)
}

// Once 一次性任务
func Once() {
func Once(isMaster bool) {
    if !isMaster {
        return
    }
    msg := &common.MsgPullDataRequest{DataType: common.PullDataTypeProcessModel}
    caller := nsq.NewCaller(fmt.Sprintf(constvar.NsqTopicPullDataRequest, conf.Conf.NsqConf.NodeId), constvar.NsqTopicPullDataResponse)
    err := caller.Send(msg)

 main.go

@@ -50,11 +50,12 @@
            logx.Errorf("nsq Init err:%v", err)
            return
        }
        crontask.Once()
        if err := crontask.StartTask(); err != nil {
            logx.Errorf("crontab task Init err:%v", err)
            return
        }

    }
    crontask.Once(agent.ClusterStatus != "slave")
    if err := crontask.StartTask(agent.ClusterStatus != "slave"); err != nil {
        logx.Errorf("crontab task Init err:%v", err)
        return
    }

    logx.Infof("apsClient start serve...")
@@ -70,18 +71,22 @@
func serfClusterEvent(stat int) {
    switch stat {
    case serf.EventCreateCluster, serf.EventSlave2Master, serf.EventLeaveCluster:
        if err := nsq.Init(); err != nil {
        if err := nsq.Init(); err != nil { //开启nsq
            logx.Errorf("nsq Init err:%v", err)
            return
        }
        crontask.Once()
        if err := crontask.StartTask(); err != nil {
        crontask.Once(true)
        if err := crontask.RestartTask(true); err != nil { //以master方式重启task
            logx.Errorf("crontab task Init err:%v", err)
            return
        }
    case serf.EventJoinCluster, serf.EventMaster2Slave:
        nsq.Stop()
        crontask.StopTask()
        nsq.Stop() //关闭nsq
        crontask.Once(false)
        if err := crontask.RestartTask(false); err != nil { //以非master方式重启task
            logx.Errorf("crontab task Init err:%v", err)
            return
        }
    }

    logx.Infof("serf cluster event: %v", stat)

 model/production_progress.go

@@ -20,10 +20,11 @@

    ProductionProgressSearch struct {
        ProductionProgress
        Order    string
        PageNum  int
        PageSize int
        Orm      *gorm.DB
        Order      string
        PageNum    int
        PageSize   int
        Orm        *gorm.DB
        UnFinished bool
    }
)

@@ -78,6 +79,11 @@
    return slf
}

func (slf *ProductionProgressSearch) SetUnFinished() *ProductionProgressSearch {
    slf.UnFinished = true
    return slf
}

func (slf *ProductionProgressSearch) build() *gorm.DB {
    var db = slf.Orm.Model(&ProductionProgress{})

@@ -109,6 +115,10 @@
        db = db.Where("channel = ?", slf.Channel)
    }

    if slf.UnFinished {
        db = db.Where("finished_quantity <  total_quantity")
    }

    return db
}


 service/progress.go

@@ -1,6 +1,7 @@
package service

import (
    "apsClient/conf"
    "apsClient/model"
    "errors"
    "github.com/jinzhu/gorm"
@@ -54,7 +55,7 @@
    var ok bool
    progressCache, ok = ProgressCacheGet(channel)
    if !ok {
        progressCache, err = model.NewProductionProgressSearch(nil).SetChannel(channel).SetOrder("id desc").First()
        progressCache, err = model.NewProductionProgressSearch(nil).SetDeviceId(conf.Conf.System.DeviceId).SetChannel(channel).SetOrder("id desc").First()
        if err == gorm.ErrRecordNotFound {
            return nil, errors.New("progress not found")
        }
@@ -71,3 +72,9 @@
    }
    return
}

// GetProgressList 获取待同步进度工序
func (slf ProgressService) GetProgressList() (progressList []*model.ProductionProgress, err error) {
    progressList, err = model.NewProductionProgressSearch(nil).SetUnFinished().SetOrder("id desc").SetPage(1, 100).FindNotTotal()
    return
}