kpt-temporary-mqtt/module/mqtt/sub.go

package mqtt

import (
	"fmt"
	"gitee.com/xuyiping_admin/pkg/logger/zaplog"
	golangMqtt "github.com/eclipse/paho.mqtt.golang"
	"go.uber.org/zap"
	"kpt-temporary-mqtt/config"
	"kpt-temporary-mqtt/model"
	"kpt-temporary-mqtt/util"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
)

var (
	configOption     = config.Options()
	golangMqttClient golangMqtt.Client
	subMsgChan       = make(chan []byte, configOption.WorkNumber)
)

var messagePubHandler golangMqtt.MessageHandler = func(client golangMqtt.Client, msg golangMqtt.Message) {
	zaplog.Info("messagePubHandlerReceived", zap.Any("message", string(msg.Payload())), zap.Any("topic", msg.Topic()))
}

var connectHandler golangMqtt.OnConnectHandler = func(client golangMqtt.Client) {
	zaplog.Info("connectedClient", zap.Any("client", client))
	client.Subscribe(configOption.SubTopName, 1, func(client golangMqtt.Client, msg golangMqtt.Message) {
		subMsgChan <- msg.Payload()
	})
}

// 连接丢失处理
var connectLostHandler golangMqtt.ConnectionLostHandler = func(client golangMqtt.Client, err error) {
	zaplog.Error("connectLost", zap.Any("err", err.Error()), zap.Any("golangMqttClient", golangMqttClient))
	connectionRetry(client)
	zaplog.Info("connectLost", zap.Any("ConnectionRetry", "ok"), zap.Any("golangMqttClient", golangMqttClient))
}

// connectionRetry 尝试重新连接
func connectionRetry(client golangMqtt.Client) {
	for {
		token := client.Connect()
		if token.Wait() && token.Error() == nil {
			// 成功重连，更新全局客户端实例
			golangMqttClient = client
			return
		}
		zaplog.Error("ConnectionRetry", zap.Any("err", token.Error()))
		time.Sleep(5 * time.Second)
	}
}

func (d *DataEventEntry) NewMqtt() {
	options := config.Options()
	opts := golangMqtt.NewClientOptions().
		AddBroker(fmt.Sprintf("tcp://%s:%d", options.Broker, options.Port)).
		SetClientID(util.RandString(12)).
		SetUsername(options.UserName).
		SetPassword(options.Password).
		SetDefaultPublishHandler(messagePubHandler)
	opts.SetMaxReconnectInterval(5 * time.Minute)
	opts.SetKeepAlive(2 * time.Minute)
	opts.SetAutoReconnect(true)
	opts.SetConnectRetry(true)
	opts.SetCleanSession(false)
	opts.OnConnect = connectHandler
	opts.OnConnectionLost = connectLostHandler

	newClient := golangMqtt.NewClient(opts)
	if token := newClient.Connect(); token.Wait() && token.Error() == nil {
		// 成功重连，更新全局客户端实例
		golangMqttClient = newClient
	} else {
		panic(token.Error())
	}
}

type DataInsertSubMsgLog struct {
	SubMsgLogList []*model.SubMsgLog
	Mx            *sync.RWMutex
}

func (d *DataEventEntry) SubMsgLog() {
	DSMLog := DataInsertSubMsgLog{
		SubMsgLogList: make([]*model.SubMsgLog, 0),
		Mx:            &sync.RWMutex{},
	}

	batchSize := 5
	batchList := make([]*model.SubMsgLog, 0, batchSize)

	sc := make(chan os.Signal, 1)
	signal.Notify(sc, os.Kill, os.Interrupt, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)

	// 设置2分钟超时
	tc := time.After(2 * time.Minute)
	for {
		select {
		case msg := <-subMsgChan:
			subMsLog := d.CreatMsgLog(msg)
			if subMsLog != nil {
				batchList = append(batchList, subMsLog)
			} else {
				zaplog.Error("subMsgChan-imei", zap.Any("subMsLog", string(msg)))
			}
			if len(batchList) >= batchSize {
				DSMLog.Mx.Lock()
				DSMLog.SubMsgLogList = append(DSMLog.SubMsgLogList, batchList...)
				if len(DSMLog.SubMsgLogList) >= 10 {
					if err := d.DB.Create(DSMLog.SubMsgLogList).Error; err != nil {
						zaplog.Error("subMsgChan-info", zap.Any("err", err), zap.Any("subMsgLog", DSMLog.SubMsgLogList))
					}
					DSMLog.SubMsgLogList = make([]*model.SubMsgLog, 0)
				}
				DSMLog.Mx.Unlock()
				batchList = batchList[:0]
			}
		// 优雅退出
		case <-sc:
			d.handleSignal(&DSMLog, subMsgChan, true)
		case <-tc:
			d.handleSignal(&DSMLog, subMsgChan, false)
		}
	}
}

var isDelete bool

func (d *DataEventEntry) CreatMsgLog(msg []byte) *model.SubMsgLog {
	if len(msg) <= 0 {
		return nil
	}
	defer func() {
		if time.Now().Day()%15 == 0 && !isDelete {
			d.DB.Model(new(model.SubMsgLog)).Where("created_at < ?", time.Now().AddDate(0, 0, -15).Unix()).Delete(new(model.SubMsgLog))
			isDelete = true
			return
		}
	}()
	subMsgLog := d.MsgDataFormat(msg)
	if subMsgLog == nil {
		return nil
	}

	if subMsgLog.Imei == "" {
		return nil
	}
	return subMsgLog
}

func (d *DataEventEntry) MsgDataFormat(msg []byte) *model.SubMsgLog {
	msgData := make(map[string]interface{})
	pairs := strings.Split(util.MsgFormat(string(msg)), " ")
	for _, pair := range pairs {
		parts := strings.SplitN(pair, ":", 2)
		if len(parts) != 2 {
			continue
		}
		key, value := parts[0], parts[1]
		if len(key) == 0 {
			continue
		}
		msgData[key] = value
	}

	softVer := int64(0)
	if softVerInter, ok := msgData["SOFT_VER"]; ok {
		if softVerstr, ok := softVerInter.(string); ok {
			softVer, _ = strconv.ParseInt(softVerstr, 10, 64)
		}
	}

	if softVer <= 0 {
		if softVerInter, ok := msgData["soft_ver"]; ok {
			if softVerstr, ok := softVerInter.(string); ok {
				softVer, _ = strconv.ParseInt(softVerstr, 10, 64)
			}
		}
	}

	uuid := ""
	if uuidInter, ok := msgData["uuid"]; ok {
		if uuidStr, ok := uuidInter.(string); ok {
			uuid = uuidStr
		}
	}

	frameId := int64(0)
	if frameIdInter, ok := msgData["frameid"]; ok {
		if frameId64, ok := frameIdInter.(string); ok {
			frameId, _ = strconv.ParseInt(frameId64, 10, 64)
		}
	}
	cowId := ""
	if cowIdInter, ok := msgData["cowid"]; ok {
		if cowIdStr, ok := cowIdInter.(string); ok {
			cowId = cowIdStr
		}
	}

	csq := int64(0)
	if csqInter, ok := msgData["csq"]; ok {
		if csq32, ok := csqInter.(string); ok {
			csq, _ = strconv.ParseInt(csq32, 10, 64)
		}
	}

	temp := float64(0)
	if tempInter, ok := msgData["Temp"]; ok {
		if tempFloat, ok := tempInter.(string); ok {
			temp, _ = strconv.ParseFloat(tempFloat, 64)
		}
	}

	if temp <= 0 {
		if tempInter, ok := msgData["temp"]; ok {
			if tempFloat, ok := tempInter.(string); ok {
				temp, _ = strconv.ParseFloat(tempFloat, 64)
			}
		}
	}

	imei := ""
	if imeiInter, ok := msgData["imei"]; ok {
		if imeiStr, ok := imeiInter.(string); ok {
			imei = imeiStr
		}
	}

	active := int64(0)
	if activeInter, ok := msgData["active"]; ok {
		if active32, ok := activeInter.(string); ok {
			active, _ = strconv.ParseInt(active32, 10, 64)
		}
	}

	inAction := int64(0)
	if inActionInter, ok := msgData["inactive"]; ok {
		if inAction32, ok := inActionInter.(string); ok {
			inAction, _ = strconv.ParseInt(inAction32, 10, 64)
		}
	}

	ruMina := int64(0)
	if ruMinaInter, ok := msgData["Rumina"]; ok {
		if ruMina32, ok := ruMinaInter.(string); ok {
			ruMina, _ = strconv.ParseInt(ruMina32, 10, 64)
		}
	}
	if ruMina <= 0 {
		if ruMinaInter, ok := msgData["rumina"]; ok {
			if ruMina32, ok := ruMinaInter.(string); ok {
				ruMina, _ = strconv.ParseInt(ruMina32, 10, 64)
			}
		}
	}

	intake := int64(0)
	if intakeInter, ok := msgData["Intake"]; ok {
		if intake32, ok := intakeInter.(string); ok {
			intake, _ = strconv.ParseInt(intake32, 10, 64)
		}
	}

	if intakeInter, ok := msgData["intake"]; ok {
		if intake32, ok := intakeInter.(string); ok {
			intake, _ = strconv.ParseInt(intake32, 10, 64)
		}
	}

	gasp := int64(0)
	if gaspInter, ok := msgData["gasp"]; ok {
		if gasp32, ok := gaspInter.(string); ok {
			gasp, _ = strconv.ParseInt(gasp32, 10, 64)
		}
	}

	other := int64(0)
	if otherInter, ok := msgData["other"]; ok {
		if other32, ok := otherInter.(string); ok {
			other, _ = strconv.ParseInt(other32, 10, 64)
		}
	}

	reMain := int64(0)
	if reMainInter, ok := msgData["Remain"]; ok {
		if reMain32, ok := reMainInter.(string); ok {
			reMain, _ = strconv.ParseInt(reMain32, 10, 64)
		}
	}

	if reMainInter, ok := msgData["remain"]; ok {
		if reMain32, ok := reMainInter.(string); ok {
			reMain, _ = strconv.ParseInt(reMain32, 10, 64)
		}
	}

	nccId := ""
	if nccIdInter, ok := msgData["nccid"]; ok {
		if nccIdStr, ok := nccIdInter.(string); ok {
			nccId = nccIdStr
		}
	}

	return &model.SubMsgLog{
		SoftVer:  softVer,
		Uuid:     uuid,
		FrameId:  frameId,
		CowId:    cowId,
		Csq:      csq,
		Temp:     int64(temp * 100),
		Imei:     imei,
		Active:   int32(active),
		InActive: int32(inAction),
		RuMina:   int32(ruMina),
		Intake:   int32(intake),
		Gasp:     int32(gasp),
		Other:    int32(other),
		ReMain:   int32(reMain),
		Nccid:    nccId,
	}
}

func (d *DataEventEntry) handleSignal(dsmLog *DataInsertSubMsgLog, subMsgChan chan []byte, isCloseChan bool) {
	if len(dsmLog.SubMsgLogList) > 0 {
		dsmLog.Mx.Lock()
		if err := d.DB.Create(dsmLog.SubMsgLogList).Error; err != nil {
			zaplog.Error("handleSignal", zap.Error(err))
		}
		zaplog.Info("handleSignal", zap.Any("success", dsmLog.SubMsgLogList))
		dsmLog.SubMsgLogList = make([]*model.SubMsgLog, 0)
		dsmLog.Mx.Unlock()
	}
	if isCloseChan {
		close(subMsgChan)
		golangMqttClient.Disconnect(250)
	}
}