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
| package collector
|
| import (
| "context"
| "sync"
| "time"
|
| "plc-recorder/logger"
| "plc-recorder/msg"
|
| plc4go "github.com/apache/plc4x/plc4go/pkg/api"
| )
|
| var mapTask sync.Map
|
| type collectorProc struct {
| device *msg.PLCDevice
| cancel context.CancelFunc
| }
|
| func InitTask() {
| device := msg.PLCDevice{
| Id: "0",
| Ip: "192.168.1.188",
| Address: []int{17021},
| Interval: 1,
| }
|
| ctx, cancel := context.WithCancel(context.Background())
| proc := collectorProc{
| device: &device,
| cancel: cancel,
| }
|
| mapTask.Store(device.Id, &proc)
|
| go connectingDevice(ctx, &device)
| }
|
| func connectingDevice(ctx context.Context, dev *msg.PLCDevice) {
| plcResponse := msg.PLCResponse{
| Id: dev.Id,
| Name: dev.Name,
| Ip: dev.Ip,
| Online: false,
| }
|
| for {
| select {
| case <-ctx.Done():
| logger.Warn("plc device %s, ip: %s, end of connecting.", dev.Name, dev.Ip)
| return
| default:
| plcConnection, err := NewModbusConnection(dev.Ip)
| if err != nil {
| logger.Warn("error connecting to PLC: %s, ip: %s", dev.Name, dev.Ip)
| plcResponse.Online = false
| msg.SendDeviceLiveData(&plcResponse)
| time.Sleep(30 * time.Second)
| } else {
| // 连接成功后, 开始采集数据, 会判断连接是否有效, 断开后会采集任务会退出, 继续重新尝试连接设备
| runCollectionTask(ctx, dev, plcConnection)
| }
| }
| }
| }
|
| func runCollectionTask(ctx context.Context, dev *msg.PLCDevice, conn plc4go.PlcConnection) {
| // 创建modbusTCP连接, 循环查询数据并上报
| plcResponse := msg.PLCResponse{
| Id: dev.Id,
| Name: dev.Name,
| Ip: dev.Ip,
| Online: true,
| Data: nil,
| }
|
| for {
| select {
| case <-ctx.Done():
| logger.Warn("plc device %s, ip: %s, end of collection.", dev.Name, dev.Ip)
| conn.Close()
| return
| default:
| if !conn.IsConnected() {
| logger.Warn("plc device %s, ip: %s, disconnected.", dev.Name, dev.Ip)
| return
| }
|
| // 根据设置的地址查询数据，上报
| plcResponse.Data = make(map[int][]byte, 0)
| for _, addr := range dev.Address {
| result, err := ReadHoldingRegister(conn, addr)
| if err != nil {
| logger.Warn("plc device Read Holding Register error, %s", err.Error())
| } else {
| plcResponse.Data[addr] = result
| }
| }
|
| msg.SendDeviceLiveData(&plcResponse)
| // 间隔时间
| time.Sleep(time.Duration(dev.Interval) * time.Second)
| }
| }
| }
|
|
