fix

zhangqian

2023-10-26 08e7f0e4148808ce435b213f09dc8b80eebb6613

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
package main
 
import (
    "apsClient/conf"
    "apsClient/crontask"
    "apsClient/model"
    "apsClient/nsq"
    "apsClient/pkg/logx"
    "apsClient/pkg/sqlitex"
    "apsClient/router"
    "apsClient/serf"
    "apsClient/service/plc_address"
    "fmt"
    "log"
    "net/http"
    "time"
)
 
func main() {
 
    logx.Init(conf.Conf.Log)
    defer logx.Sync()
 
    if err := model.Init(); err != nil {
        logx.Errorf("model Init err:%v", err)
        return
    }
 
    //加载plc写入地址
    plc_address.LoadAddressFromFile()
 
    // 启动数据同步
    var serfStartChan = make(chan bool)
 
    // 需要同步的表
    var syncTables = []string{
        "procedures",
        "process_model",
        "production_progress",
        "work_order",
        "task_status_sync",
        "device",
        "device_plc",
    }
 
    agent := serf.InitAgent("apsClient", syncTables, sqlitex.GetDB())
    agent.RegisterClusterEvent(serfClusterEvent)
 
    go agent.Serve(serfStartChan)
    if !<-serfStartChan {
        logx.Errorf("serf Init err, exit")
        return
    }
 
    // 判断当前集群状态
    logx.Infof("current agent.ClusterStatus:%v", agent.ClusterStatus)
    log.Println("current agent.ClusterStatus:", agent.ClusterStatus)
    if agent.ClusterStatus != "slave" {
        if err := nsq.Init(); err != nil {
            logx.Errorf("nsq 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...")
    server := &http.Server{
        Addr:         fmt.Sprintf(":%d", conf.Conf.System.Port),
        Handler:      router.InitRouter(),
        ReadTimeout:  5 * time.Second,
        WriteTimeout: 5 * time.Second,
    }
    logx.Error(server.ListenAndServe().Error())
}
 
func serfClusterEvent(stat int) {
    switch stat {
    case serf.EventCreateCluster, serf.EventSlave2Master, serf.EventLeaveCluster:
        if err := nsq.Init(); err != nil { //开启nsq
            logx.Errorf("nsq Init err:%v", err)
            return
        }
        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() //关闭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)
}