ipole_edge_server/cloud/websvr/controllers/eventmgr.go

package controllers

import (
	"fmt"
	"runtime/debug"
	"strconv"
	"sync"
	"time"

	"github.com/astaxie/beego"

	"lc/common/models"
	"lc/common/util"
)

// DeviceAlarmId 告警处理
var DeviceAlarmId = "device_alarm_id"

var _EventMgrOnce sync.Once
var _EventMgrSingle *EventMgr

func GetEventMgr() *EventMgr {
	_EventMgrOnce.Do(func() {
		_EventMgrSingle = &EventMgr{
			MapAlarm:    make(map[string]int64),
			MapLastData: make(map[string]time.Time),
		}
	})
	return _EventMgrSingle
}

type EventObject struct {
	ID    string
	Etype models.EventType
	Time  time.Time
	Value float64
}

type EventMgr struct {
	mu          sync.Mutex
	MapLastData map[string]time.Time
	MapAlarm    map[string]int64
}

func (o *EventMgr) Update(code string) {
	if code != "" {
		o.mu.Lock()
		o.MapLastData[code] = util.MlNow()
		o.mu.Unlock()
	}
}

func (o *EventMgr) Handler() {
	defer func() {
		if err := recover(); err != nil {
			time.Sleep(5 * time.Second)
			go o.Handler()
			beego.Error(fmt.Sprintf("EventMgr.Handler发生异常:%v", err))
			beego.Error("EventMgr.Handler发生异常:%s", string(debug.Stack()))
		}
	}()
	timer := time.NewTicker(1 * time.Minute)
	_mapLastData := make(map[string]time.Time)
	_mapRedis := make(map[string]bool)
	for {
		select {
		case <-timer.C: //每隔5分钟执行一次
			o.mu.Lock()
			for k, v := range o.MapLastData {
				_mapLastData[k] = v
			}
			o.mu.Unlock()

			for k, v := range _mapLastData {
				if util.MlNow().Sub(v).Minutes() > 5 { //离线
					aid, ok := o.MapAlarm[k]
					if !ok {
						//试图从redis查找，看是不是已有离线告警，有则忽略，无则处理，新建告警
						if _, ok := _mapRedis[k]; !ok {
							if str, err := redisCltRawdata.HGet(DeviceAlarmId, k).Result(); err == nil {
								if id, err := strconv.Atoi(str); err == nil {
									aid = int64(id)
								}
								_mapRedis[k] = true
							}
						}
						if aid > 0 {
							continue
						}
						oo := models.DeviceAlarm{
							DID:       k,
							TStart:    v,
							Threshold: 0,
							SValue:    0,
							Content:   "离线",
							AlarmType: 0,
							Level:     2, //设备离线为严重告警
						}
						if err := models.G_db.Create(&oo).Error; err != nil {
							beego.Error(fmt.Sprintf("告警信息[%v]入库失败:%s", oo, err.Error()))
						} else {
							o.MapAlarm[k] = oo.ID
							if err := redisCltRawdata.HSet(DeviceAlarmId, k, oo.ID).Err(); err != nil {
								beego.Error(fmt.Sprintf("设备[%s]告警数据[%d]缓存失败:%s", k, oo.ID, err.Error()))
							}
						}
						o.SaveEventObject(&EventObject{ID: k, Etype: models.ET_OFFLINE, Time: v, Value: 0})
					}
					redisCltRawdata.HMSet(DevStatusPrefix+k, map[string]interface{}{TLast: util.MlNow().Format("2006-01-02 15:04:05"), ONLINE: 0})
				} else { //在线
					aid, ok := o.MapAlarm[k]
					if !ok {
						//查找过redis，则不再从redis查找
						if _, ok := _mapRedis[k]; !ok {
							if str, err := redisCltRawdata.HGet(DeviceAlarmId, k).Result(); err == nil {
								if id, err := strconv.Atoi(str); err == nil {
									aid = int64(id)
								}
								_mapRedis[k] = true
							}
						}
					}
					if aid > 0 {
						oo := models.DeviceAlarm{ID: aid, TEnd: v, EValue: float32(0)}
						if err := oo.Update(); err != nil {
							beego.Error(fmt.Sprintf("更新告警[%d]信息失败:%s", aid, err.Error()))
						}
						delete(o.MapAlarm, k)
						if err := redisCltRawdata.HDel(DeviceAlarmId, k).Err(); err != nil {
							beego.Error(fmt.Sprintf("更新告警[%d]信息失败:%s", aid, err.Error()))
						}
						o.SaveEventObject(&EventObject{ID: k, Etype: models.ET_ONLINE, Time: v, Value: 0})
					}
					redisCltRawdata.HMSet(DevStatusPrefix+k, map[string]interface{}{TLast: util.MlNow().Format("2006-01-02 15:04:05"), ONLINE: 1})
				}
			}
		default:
			time.Sleep(time.Minute)
		}
	}
}

func (o *EventMgr) SaveEventObject(eo *EventObject) {
	oo := models.DeviceEvents{DID: eo.ID, EType: eo.Etype, Time: eo.Time}
	if err := models.G_db.Create(&oo).Error; err != nil {
		beego.Error(fmt.Sprintf("设备[%s]事件入库失败:%s", eo.ID, err.Error()))
	}
}