事件幂等性操作

zhangqian

2023-10-21 e1890392075ce73bff71c5dc0167f8af46cba0a1

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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
package service
 
import (
    "apsClient/conf"
    "apsClient/constvar"
    "apsClient/model"
    "fmt"
    "github.com/jinzhu/gorm"
    "github.com/spf13/cast"
    "sync"
    "time"
)
 
type CacheStore struct {
    cache map[string]interface{}
    mu    sync.Mutex
}
 
var defaultCacheStore *CacheStore
 
func init() {
    defaultCacheStore = newCacheManager()
}
func newCacheManager() *CacheStore {
    return &CacheStore{
        cache: make(map[string]interface{}),
    }
}
 
func (cm *CacheStore) Get(key string) (interface{}, bool) {
    cm.mu.Lock()
    defer cm.mu.Unlock()
 
    conn, ok := cm.cache[key]
    return conn, ok
}
 
func (cm *CacheStore) Add(key string, value interface{}) {
    cm.mu.Lock()
    defer cm.mu.Unlock()
    cm.cache[key] = value
}
 
func (cm *CacheStore) Remove(key string) {
    cm.mu.Lock()
    defer cm.mu.Unlock()
    delete(cm.cache, key)
}
 
const (
    PlcCacheKey             = "plc:%v:%v"                    //plc:channel:key
    CurrentTaskCacheKey     = "current_task:%v"              //current_task:channel
    CurrentProgressCacheKey = "current_progress:%v"          //current_progress:channel
    PlcCacheKeyUpdateTime   = "finish_number_update_time:%v" //finish_number_update_time:channel
)
 
func PlcCacheGet(channel int32, key string) (interface{}, bool) {
    return defaultCacheStore.Get(fmt.Sprintf(PlcCacheKey, channel, key))
}
 
func PlcCacheSet(channel int32, key string, value interface{}) {
    if key == constvar.PlcCacheKeyFinishNumber {
        oldFinishNumber, exists := PlcCacheGet(channel, key)
        if !exists || cast.ToInt(oldFinishNumber) != cast.ToInt(value) { //finishNumber有了变化，设置更新时间缓存
            FinishUpdateTimeSet(channel, time.Now().Unix())
        }
    }
    defaultCacheStore.Add(fmt.Sprintf(PlcCacheKey, channel, key), value)
}
 
func FinishUpdateTimeGet(channel int32) interface{} {
    val, ok := defaultCacheStore.Get(fmt.Sprintf(PlcCacheKeyUpdateTime, channel))
    if ok {
        return val
    }
    return 0
}
 
func FinishUpdateTimeSet(channel int32, value interface{}) {
    defaultCacheStore.Add(fmt.Sprintf(PlcCacheKeyUpdateTime, channel), value)
}
 
func TaskFlagSet(channel int32) {
    defaultCacheStore.Add(fmt.Sprintf(CurrentTaskCacheKey, channel), true)
}
 
func TaskFlagUnset(channel int32) {
    defaultCacheStore.Add(fmt.Sprintf(CurrentTaskCacheKey, channel), false)
}
 
func TaskFlagGet(channel int32) bool {
    if v, ok := defaultCacheStore.Get(fmt.Sprintf(CurrentTaskCacheKey, channel)); ok {
        return v.(bool)
    }
    _, err := model.NewProceduresSearch(nil).SetDeviceId(conf.Conf.System.DeviceId).SetStatus(model.ProcedureStatusProcessing).SetChannels([]int32{channel}).First()
    if err == gorm.ErrRecordNotFound {
        defaultCacheStore.Add(fmt.Sprintf(CurrentTaskCacheKey, channel), false)
        return false
    } else {
        defaultCacheStore.Add(fmt.Sprintf(CurrentTaskCacheKey, channel), true)
        return true
    }
}
 
func ProgressCacheGet(channel int32) (*model.ProductionProgress, bool) {
    if v, ok := defaultCacheStore.Get(fmt.Sprintf(CurrentProgressCacheKey, channel)); ok {
        return v.(*model.ProductionProgress), ok
    }
    return nil, false
}
 
func ProgressCacheSet(channel int32, value *model.ProductionProgress) {
    defaultCacheStore.Add(fmt.Sprintf(CurrentProgressCacheKey, channel), value)
}
 
func ProgressCacheUnset(channel int32) {
    defaultCacheStore.Remove(fmt.Sprintf(CurrentProgressCacheKey, channel))
}