mqtt: 优化

config/app.develop.yaml

@@ -5,6 +5,7 @@ debug: true
 http_server_addr: ':8090'
 http_metrics_addr: ':23332'
 jwt_expire_time: 172800
+neck_ring_limit: 10000
 
 store:
   show_sql: true

config/app.go

@@ -24,8 +24,9 @@ type AppConfig struct {
 	FarmName       string `yaml:"farm_name"`
 	AppName        string `yaml:"app_name"`
 	AppEnv         string `yaml:"app_environment"`
-	Debug          bool   `yaml:"debug" env:"APP_DEBUG"`
-	HTTPServerAddr string `yaml:"http_server_addr" env:"HTTP_SERVER_ADDR"`
+	Debug          bool   `yaml:"debug"`
+	HTTPServerAddr string `yaml:"http_server_addr"`
+	NeckRingLimit  int32  `yaml:"neck_ring_limit"`
 
 	// 数据库配置 额外加载文件部分 database.yaml
 	StoreSetting StoreSetting `json:"storeSetting" yaml:"store"`

config/app.test.yaml

@@ -4,6 +4,7 @@ debug: true
 http_server_addr: ':8090'
 http_metrics_addr: ':23332'
 jwt_expire_time: 172800
+neck_ring_limit: 10000
 
 store:
   show_sql: true

dep/dep.go

@@ -7,6 +7,7 @@ import (
 	"kpt-pasture/module/crontab"
 	"kpt-pasture/service/asynqsvc"
 	"kpt-pasture/service/mqtt"
+	"kpt-pasture/service/mqtt2"
 	"kpt-pasture/service/redis"
 	"kpt-pasture/service/sso"
 	"kpt-pasture/service/wechat"
@@ -38,5 +39,6 @@ func Options() []di.HubOption {
 		redis.Module,
 		crontab.Module,
 		mqtt.Module,
+		mqtt2.Module,
 	}
 }

go.mod

@@ -3,7 +3,7 @@ module kpt-pasture
 go 1.17
 
 require (
-	gitee.com/xuyiping_admin/go_proto v0.0.0-20241127090200-9343aa730c3f
+	gitee.com/xuyiping_admin/go_proto v0.0.0-20241128094754-04053736c8ab
 	gitee.com/xuyiping_admin/pkg v0.0.0-20241108060137-caea58c59f5b
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/eclipse/paho.mqtt.golang v1.4.3

go.sum

@@ -52,6 +52,8 @@ gitee.com/xuyiping_admin/go_proto v0.0.0-20241127084727-32ff0826e19a h1:JDS3rqbw
 gitee.com/xuyiping_admin/go_proto v0.0.0-20241127084727-32ff0826e19a/go.mod h1:BKrFW6YLDectlQcQk3FYKBeXvjEiodAKJ5rq7O/QiPE=
 gitee.com/xuyiping_admin/go_proto v0.0.0-20241127090200-9343aa730c3f h1:iRYyVkDAbWJG56FUqvbuNkMgDU7v+fkCI2iuepaNxEw=
 gitee.com/xuyiping_admin/go_proto v0.0.0-20241127090200-9343aa730c3f/go.mod h1:BKrFW6YLDectlQcQk3FYKBeXvjEiodAKJ5rq7O/QiPE=
+gitee.com/xuyiping_admin/go_proto v0.0.0-20241128094754-04053736c8ab h1:9pYdwjsGBoT8frMz42gDq4muVRkGJuidkaMmLVgpICg=
+gitee.com/xuyiping_admin/go_proto v0.0.0-20241128094754-04053736c8ab/go.mod h1:BKrFW6YLDectlQcQk3FYKBeXvjEiodAKJ5rq7O/QiPE=
 gitee.com/xuyiping_admin/pkg v0.0.0-20241108060137-caea58c59f5b h1:w05MxH7yqveRlaRbxHhbif5YjPrJFodRPfOjYhXn7Zk=
 gitee.com/xuyiping_admin/pkg v0.0.0-20241108060137-caea58c59f5b/go.mod h1:8tF25X6pE9WkFCczlNAC0K2mrjwKvhhp02I7o0HtDxY=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=

http/handler/event/event_base.go

@@ -174,7 +174,7 @@ func BodyScoreEventCreate(c *gin.Context) {
 	})
 }
 
-func WeightEventList(c *gin.Context) {
+func WeightList(c *gin.Context) {
 	var req pasturePb.SearchEventRequest
 	if err := ginutil.BindProto(c, &req); err != nil {
 		apierr.AbortBadRequest(c, http.StatusBadRequest, err)
@@ -195,7 +195,7 @@ func WeightEventList(c *gin.Context) {
 	ginutil.JSONResp(c, res)
 }
 
-func WeightEventCreate(c *gin.Context) {
+func WeightBatch(c *gin.Context) {
 	var req pasturePb.EventWeight
 	if err := ginutil.BindProto(c, &req); err != nil {
 		apierr.AbortBadRequest(c, http.StatusBadRequest, err)
@@ -203,13 +203,13 @@ func WeightEventCreate(c *gin.Context) {
 	}
 
 	if err := valid.ValidateStruct(&req,
-		valid.Field(&req.CowId, valid.Required),
 		valid.Field(&req.WeightAt, valid.Required),
 		valid.Field(&req.OperationId, valid.Required),
 		valid.Field(&req.WeightItems, valid.Required, valid.Each(valid.By(func(value interface{}) error {
 			s := value.(pasturePb.WeightItem)
 			return valid.ValidateStruct(&s,
 				valid.Field(&s.Weight, valid.Required),
+				valid.Field(&s.CowId, valid.Required),
 			)
 		}))),
 	); err != nil {
@@ -217,7 +217,7 @@ func WeightEventCreate(c *gin.Context) {
 		return
 	}
 
-	if err := middleware.BackendOperation(c).OpsService.WeightCreate(c, &req); err != nil {
+	if err := middleware.BackendOperation(c).OpsService.WeightBatch(c, &req); err != nil {
 		apierr.ClassifiedAbort(c, err)
 		return
 	}

http/route/event_api.go

@@ -25,15 +25,15 @@ func EventAPI(opts ...func(engine *gin.Engine)) func(s *gin.Engine) {
 		eventRoute.POST("/body/score/create", event.BodyScoreEventCreate)
 
 		// 称重
-		eventRoute.POST("/weight/list", event.WeightEventList)
-		eventRoute.POST("/weight/create", event.WeightEventCreate)
+		eventRoute.POST("/weight/list", event.WeightList)
+		eventRoute.POST("/weight/batch", event.WeightBatch)
 
 		// 产犊
 		eventRoute.POST("/calving/list", event.CalvingEventList)
 		eventRoute.POST("/calving/create", event.CalvingEventCreate)
 		// 孕检
 		eventRoute.POST("/pregnant/check/list", event.PregnantCheckEventList)
-		eventRoute.POST("/pregnant/check/create/batch", event.PregnantCheckEventCreateBatch)
+		eventRoute.POST("/pregnant/check/batch", event.PregnantCheckEventCreateBatch)
 		// 配种
 		eventRoute.POST("/mating/list", event.MatingEventList)
 		eventRoute.POST("/mating/create", event.MatingCreate)

module/backend/calendar.go

@@ -225,7 +225,7 @@ func (s *StoreEntry) SameTimeCowList(ctx context.Context, req *pasturePb.ItemsRe
 		pref.Where("a.same_time_type = ?", req.CalendarType)
 	}
 
-	if req.CowType >= 0 {
+	if req.CowType > 0 {
 		pref.Where("b.cow_type = ?", req.CowType)
 	}
 
@@ -294,8 +294,8 @@ func (s *StoreEntry) PregnancyCheckCowList(ctx context.Context, req *pasturePb.I
 		pref.Where("a.status = ?", req.Status)
 	}
 
-	if req.PregnantCheckName > 0 {
-		pref.Where("pregnant_check_name = ?", model.PregnantCheckNameValueMap[req.PregnantCheckName])
+	if req.PregnantCheckType > 0 {
+		pref.Where("pregnant_check_name = ?", model.PregnantCheckNameValueMap[req.PregnantCheckType])
 	}
 
 	if err := pref.Order("id desc").

module/backend/event_base.go

@@ -272,8 +272,8 @@ func (s *StoreEntry) WeightList(ctx context.Context, req *pasturePb.SearchEventR
 	}, nil
 }
 
-func (s *StoreEntry) WeightCreate(ctx context.Context, req *pasturePb.EventWeight) error {
-	if len(req.CowId) <= 0 {
+func (s *StoreEntry) WeightBatch(ctx context.Context, req *pasturePb.EventWeight) error {
+	if len(req.WeightItems) <= 0 {
 		return xerr.Custom("请选择相关牛只")
 	}
 	currentUser, err := s.GetCurrentSystemUser(ctx)
@@ -302,10 +302,11 @@ func (s *StoreEntry) WeightCreate(ctx context.Context, req *pasturePb.EventWeigh
 
 	if err = s.DB.Transaction(func(tx *gorm.DB) error {
 		for _, item := range weightEvent {
-			if err = tx.Model(new(model.Cow)).Where("id = ?", item.CowId).Updates(map[string]interface{}{
-				"last_weight_at": item.WeightAt,
-				"current_weight": item.Weight,
-			}).Error; err != nil {
+			if err = tx.Model(new(model.Cow)).Where("id = ?", item.CowId).
+				Updates(map[string]interface{}{
+					"last_weight_at": item.WeightAt,
+					"current_weight": item.Weight,
+				}).Error; err != nil {
 				return xerr.WithStack(err)
 			}
 		}

module/backend/event_breed.go

@@ -288,7 +288,7 @@ func (s *StoreEntry) SameTimeBatch(ctx context.Context, req *pasturePb.EventSame
 				Where("status = ?", pasturePb.IsShow_No).
 				Updates(map[string]interface{}{
 					"status":         pasturePb.IsShow_Ok,
-					"drug_id":        drugs.Id,
+					"drugs_id":       drugs.Id,
 					"unit":           drugs.Unit,
 					"usage":          req.Usage,
 					"remarks":        req.Remarks,

module/backend/interface.go

@@ -181,7 +181,7 @@ type EventService interface {
 	SameTimeList(ctx context.Context, req *pasturePb.SearchEventRequest, pagination *pasturePb.PaginationModel) (*pasturePb.SearchSameTimeResponse, error)
 	// WeightList 称重
 	WeightList(ctx context.Context, req *pasturePb.SearchEventRequest, pagination *pasturePb.PaginationModel) (*pasturePb.SearchWeightEventResponse, error)
-	WeightCreate(ctx context.Context, req *pasturePb.EventWeight) error
+	WeightBatch(ctx context.Context, req *pasturePb.EventWeight) error
 
 	// CowDiseaseCreate 提交发病牛只
 	CowDiseaseCreate(ctx context.Context, req *pasturePb.EventCowDiseaseRequest) error

module/crontab/neck_ring.go

@@ -0,0 +1,49 @@
+package crontab
+
+import (
+	"kpt-pasture/config"
+	"kpt-pasture/model"
+
+	pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow"
+	"gitee.com/xuyiping_admin/pkg/xerr"
+)
+
+func (e *Entry) NeckRingData() error {
+	cfg := config.Options()
+	limit := cfg.NeckRingLimit
+	if limit <= 0 {
+		limit = 10000
+	}
+	neckRingList := make([]*model.NeckRingOriginalData, 0)
+	if err := e.DB.Model(new(model.NeckRingOriginalData)).
+		Where("is_show = ?", pasturePb.IsShow_No).
+		Order("id asc").Limit(10000).Find(&neckRingList).Error; err != nil {
+		return xerr.WithStack(err)
+	}
+
+	if len(neckRingList) <= 0 {
+		return nil
+	}
+
+	neckRingIds := make([]int64, len(neckRingList))
+	// 更新已处理过的id
+	defer func() {
+		if len(neckRingIds) > 0 {
+			e.DB.Model(new(model.NeckRingOriginalData)).
+				Where("id IN ?", neckRingIds).
+				Update("is_show", pasturePb.IsShow_Ok)
+		}
+	}()
+	for i, v := range neckRingList {
+		neckRingIds[i] = v.Id
+		if v.Imei == "" {
+			continue
+		}
+
+	}
+	return nil
+}
+
+func (e *Entry) NeckRingOriginalDataMerge() {
+
+}

module/crontab/work_cron.go

@@ -154,32 +154,3 @@ func (e *Entry) CowPregnant() error {
 	}
 	return nil
 }
-
-func (e *Entry) NeckRingData() error {
-	neckRingList := make([]*model.NeckRingOriginalData, 0)
-	if err := e.DB.Model(new(model.NeckRingOriginalData)).
-		Where("is_show = ?", pasturePb.IsShow_No).
-		Find(&neckRingList).Error; err != nil {
-		return xerr.WithStack(err)
-	}
-
-	if len(neckRingList) <= 0 {
-		return nil
-	}
-
-	neckRingIds := make([]int64, len(neckRingList))
-	// 更新已处理过的id
-	defer func() {
-		if len(neckRingIds) > 0 {
-			e.DB.Model(new(model.NeckRingOriginalData)).Where("id IN ?", neckRingIds).Update("is_show", pasturePb.IsShow_Ok)
-		}
-	}()
-	for i, v := range neckRingList {
-		neckRingIds[i] = v.Id
-		if v.Imei == "" {
-			continue
-		}
-
-	}
-	return nil
-}

service/mqtt2/mqtt.go

@@ -0,0 +1,67 @@
+package mqtt2
+
+import (
+	"fmt"
+	"kpt-pasture/config"
+
+	"gitee.com/xuyiping_admin/pkg/di"
+	"gitee.com/xuyiping_admin/pkg/logger/zaplog"
+	golangMqtt "github.com/eclipse/paho.mqtt.golang"
+	"go.uber.org/zap"
+)
+
+var Module = di.Options(
+	di.Provide(NewServer),
+)
+
+type Server interface {
+	Subscribe(topic string, qos int32) []byte
+	Publish(topic string, qos int32, retained bool, payload string) error
+}
+
+type MqttServer struct {
+	client golangMqtt.Client
+	Conf   config.MqttSetting
+}
+
+func NewServer(conf config.MqttSetting) *MqttServer {
+	opts := golangMqtt.NewClientOptions()
+	opts.AddBroker(fmt.Sprintf("tcp://%s:%d", conf.Broker, conf.Port))
+	opts.SetClientID(conf.ClientId)
+	opts.SetCleanSession(false)
+	opts.SetUsername(conf.UserName)
+	opts.SetPassword(conf.Password)
+	opts.SetAutoReconnect(conf.AutoReconnect)
+	opts.SetDefaultPublishHandler(messagePubHandler)
+	opts.OnConnect = connectHandler
+	opts.OnConnectionLost = connectLostHandler
+	client := golangMqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		panic(token.Error())
+	}
+
+	return &MqttServer{client: client, Conf: conf}
+}
+
+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))
+}
+
+var connectLostHandler golangMqtt.ConnectionLostHandler = func(client golangMqtt.Client, err error) {
+	zaplog.Info("connectLost", zap.Any("err", err.Error()))
+}
+
+func (s *MqttServer) Subscribe(topic string, qos int32, handler golangMqtt.MessageHandler) {
+	if token := s.client.Subscribe(topic, byte(qos), handler); token.Wait() && token.Error() != nil {
+		panic(token.Error())
+	}
+}
+
+func (s *MqttServer) Publish(topic string, qos int32, retained bool, payload string) {
+	token := s.client.Publish(topic, byte(qos), retained, payload)
+	token.Wait()
+}

service/mqtt2/sub.go

@@ -0,0 +1,31 @@
+package mqtt2
+
+import (
+	"fmt"
+	"sync"
+
+	golangMqtt "github.com/eclipse/paho.mqtt.golang"
+
+	"go.uber.org/dig"
+)
+
+type MQTTConsumer struct {
+	dig.In
+
+	MQTTServer *MqttServer
+}
+
+var bufferPool = sync.Pool{
+	New: func() interface{} {
+		return make([]byte, 1024) // 根据实际情况调整缓冲区大小
+	},
+}
+
+func (c *MQTTConsumer) Start() {
+
+	handler := func(client golangMqtt.Client, msg golangMqtt.Message) {
+		fmt.Printf("MQTT message received: %s\n", msg.Payload())
+		// 处理消息的逻辑
+	}
+	c.MQTTServer.Subscribe(c.MQTTServer.Conf.Topic, int32(c.MQTTServer.Conf.Qos), handler)
+}