refactor:优化日志记录,mqtt重连修改

yangfu
main.go
pkg/application/crontab/crontab_auto_approve_attendance_record.go
pkg/application/crontab/crontab_auto_approve_product_record.go
pkg/application/crontab/crontab_auto_fresh_device_daily_running_record.go
pkg/application/crontab/crontab_auto_fresh_device_daily_running_record_oee.go
pkg/application/crontab/crontab_auto_workshop_plan_completion_record.go
pkg/application/crontab/crontab_sync_product.go
pkg/application/crontab/crontab_sync_product_plan.go
pkg/infrastructure/mqtt/subscribe_client.go
pkg/log/logrus.go
pkg/port/mqtt/mqtt.go
--- a/main.go
查看文件 @696a895
+++ b/main.go
查看文件 @696a895
@@ -39,7 +39,7 @@ func main() {
 	log.Logger.Info("server start ....")
 	log.Logger.Debug(fmt.Sprintf("ENABLE_KAFKA_LOG:%v", constant.ENABLE_KAFKA_LOG))
-	go mqtt.Start()
+	go mqtt.Start(log.Logger)
 	go task.Run()
 	cron := crontab.NewCrontabService(nil)
 	cron.StartCrontabTask()
--- a/pkg/application/crontab/crontab_auto_approve_attendance_record.go
查看文件 @696a895
+++ b/pkg/application/crontab/crontab_auto_approve_attendance_record.go
查看文件 @696a895
@@ -15,7 +15,7 @@ import (
 func AutoApproveProductAttendanceRecord(ctx context.Context) error {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Logger.Error(fmt.Sprintf("%v", r))
+			log.Logger.Error(fmt.Sprintf("%v", r), map[string]interface{}{"task": "定时审核生产考勤记录"})
 		}
 	}()
 	transactionContext, err := factory.CreateTransactionContext(nil)
@@ -36,12 +36,14 @@ func AutoApproveProductAttendanceRecord(ctx context.Context) error {
 	var records []*domain.ProductAttendanceRecord
 	_, records, err = attendanceRecordDao.RecentUnApprovedAttendanceRecord(24, 2)
 	if err != nil {
+		log.Logger.Error(err.Error(), map[string]interface{}{"task": "定时审核生产考勤记录"})
 		return err
 	}
 	approveAttendanceRecordsService, _ := domainService.NewPGApproveAttendanceRecordsService(transactionContext.(*pgTransaction.TransactionContext))
 	if _, err = approveAttendanceRecordsService.BatchApproveAttendanceRecords(nil, records, 0, 0, domain.AttendanceAutoApproved); err != nil {
+		log.Logger.Error(err.Error(), map[string]interface{}{"task": "定时审核生产考勤记录"})
 		return err
 	}
 	if err = transactionContext.CommitTransaction(); err != nil {
--- a/pkg/application/crontab/crontab_auto_approve_product_record.go
查看文件 @696a895
+++ b/pkg/application/crontab/crontab_auto_approve_product_record.go
查看文件 @696a895
@@ -15,7 +15,7 @@ import (
 func AutoApproveProductRecord(ctx context.Context) error {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Logger.Error(fmt.Sprintf("%v", r))
+			log.Logger.Error(fmt.Sprintf("%v", r), map[string]interface{}{"task": "定时审核生产记录"})
 		}
 	}()
 	transactionContext, err := factory.CreateTransactionContext(nil)
@@ -42,6 +42,7 @@ func AutoApproveProductRecord(ctx context.Context) error {
 	approveAttendanceRecordsService, _ := domainService.NewPGProductRecordService(transactionContext.(*pgTransaction.TransactionContext))
 	if _, err = approveAttendanceRecordsService.BatchApprove(records, 0, 0, domain.AttendanceAutoApproved); err != nil {
+		log.Logger.Error(err.Error(), map[string]interface{}{"task": "定时审核生产记录"})
 		return err
 	}
 	if err = transactionContext.CommitTransaction(); err != nil {
--- a/pkg/application/crontab/crontab_auto_fresh_device_daily_running_record.go
查看文件 @696a895
+++ b/pkg/application/crontab/crontab_auto_fresh_device_daily_running_record.go
查看文件 @696a895
@@ -13,7 +13,7 @@ import (
 func AutoFlushDeviceDailyRunningRecord(ctx context.Context) error {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Logger.Error(fmt.Sprintf("%v", r))
+			log.Logger.Error(fmt.Sprintf("%v", r), map[string]interface{}{"task": "定时刷新设备每日运行记录"})
 		}
 	}()
 	transactionContext, err := factory.CreateTransactionContext(nil)
--- a/pkg/application/crontab/crontab_auto_fresh_device_daily_running_record_oee.go
查看文件 @696a895
+++ b/pkg/application/crontab/crontab_auto_fresh_device_daily_running_record_oee.go
查看文件 @696a895
@@ -18,7 +18,7 @@ import (
 func AutoFlushDeviceDailyRunningRecordOEE(ctx context.Context) error {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Logger.Error(fmt.Sprintf("%v", r))
+			log.Logger.Error(fmt.Sprintf("%v", r), map[string]interface{}{"task": "定时刷新设备每日运行记录"})
 		}
 	}()
 	transactionContext, err := factory.CreateTransactionContext(nil)
@@ -92,7 +92,7 @@ func AutoFlushDeviceDailyRunningRecordOEE(ctx context.Context) error {
 			continue
 		}
 		if err := redis.SaveDeviceDailyRunningRecord(v); err != nil {
-			log.Logger.Error(err.Error())
+			log.Logger.Error(err.Error(), map[string]interface{}{"task": "定时刷新设备每日运行记录"})
 			continue
 		}
 		log.Logger.Debug(fmt.Sprintf("【定时刷新设备每日运行记录-OEE】 刷新记录 %v", v))
--- a/pkg/application/crontab/crontab_auto_workshop_plan_completion_record.go
查看文件 @696a895
+++ b/pkg/application/crontab/crontab_auto_workshop_plan_completion_record.go
查看文件 @696a895
@@ -15,7 +15,7 @@ import (
 func AutoWorkshopPlanCompletionRecord(ctx context.Context) error {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Logger.Error(fmt.Sprintf("%v", r))
+			log.Logger.Error(fmt.Sprintf("%v", r), map[string]interface{}{"task": "定时刷新车间计划完成纪录"})
 		}
 	}()
 	transactionContext, err := factory.CreateTransactionContext(nil)
@@ -38,6 +38,7 @@ func AutoWorkshopPlanCompletionRecord(ctx context.Context) error {
 	approveAttendanceRecordsService, _ := domainService.NewPGWorkshopPlanCompletionRecordService(transactionContext.(*pgTransaction.TransactionContext))
 	if err = approveAttendanceRecordsService.WorkshopPlanCompletion(begin, end); err != nil {
+		log.Logger.Error(err.Error(), map[string]interface{}{"task": "定时刷新车间计划完成纪录"})
 		return err
 	}
@@ -74,6 +75,7 @@ func AutoTodayWorkshopPlanCompletionRecord(ctx context.Context) error {
 	approveAttendanceRecordsService, _ := domainService.NewPGWorkshopPlanCompletionRecordService(transactionContext.(*pgTransaction.TransactionContext))
 	if err = approveAttendanceRecordsService.WorkshopPlanCompletion(begin, end); err != nil {
+		log.Logger.Error(err.Error(), map[string]interface{}{"task": "定时刷新当天车间计划完成纪录"})
 		return err
 	}
--- a/pkg/application/crontab/crontab_sync_product.go
查看文件 @696a895
+++ b/pkg/application/crontab/crontab_sync_product.go
查看文件 @696a895
@@ -8,6 +8,7 @@ import (
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/syncdata"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/redis"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
+	"runtime/debug"
 	"time"
 )
@@ -15,7 +16,7 @@ import (
 func SyncProduct(ctx context.Context) error {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Logger.Error(fmt.Sprintf("%v", r))
+			log.Logger.Error(fmt.Sprintf("%v", r), map[string]interface{}{"task": "定时同步产品", "stack": string(debug.Stack())})
 		}
 	}()
 	transactionContext, err := factory.CreateTransactionContext(nil)
@@ -36,9 +37,11 @@ func SyncProduct(ctx context.Context) error {
 	pullK3CloudService := syncdata.PullDataK3CloudService{}
 	t := ctx.Value("fromTime")
 	var fromTime time.Time
-	if t != nil {
-		if v, ok := t.(time.Time); ok {
-			fromTime = v
+	if ctx != nil {
+		if t != nil {
+			if v, ok := t.(time.Time); ok {
+				fromTime = v
+			}
 		}
 	}
 	if fromTime.IsZero() {
@@ -49,7 +52,7 @@ func SyncProduct(ctx context.Context) error {
 		}
 	}
 	if err := pullK3CloudService.SyncDataProduct(transactionContext.(*pg.TransactionContext), fromTime); err != nil {
-		log.Logger.Error(err.Error())
+		log.Logger.Error(err.Error(), map[string]interface{}{"task": "定时同步产品"})
 		return nil
 	}
 	if err = transactionContext.CommitTransaction(); err != nil {
--- a/pkg/application/crontab/crontab_sync_product_plan.go
查看文件 @696a895
+++ b/pkg/application/crontab/crontab_sync_product_plan.go
查看文件 @696a895
@@ -7,6 +7,7 @@ import (
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/factory"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/syncdata"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
+	"runtime/debug"
 	"time"
 )
@@ -14,7 +15,7 @@ import (
 func SyncProductPlan(ctx context.Context) error {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Logger.Error(fmt.Sprintf("%v", r))
+			log.Logger.Error(fmt.Sprintf("%v", r), map[string]interface{}{"task": "定时同步车间计划", "stack": debug.Stack()})
 		}
 	}()
 	transactionContext, err := factory.CreateTransactionContext(nil)
@@ -33,15 +34,18 @@ func SyncProductPlan(ctx context.Context) error {
 	log.Logger.Debug("【定时同步车间计划】 启动")
 	pullK3CloudService := syncdata.PullDataK3CloudService{}
-	t := ctx.Value("fromTime")
 	var fromTime time.Time
-	if t != nil {
-		if v, ok := t.(time.Time); ok {
-			fromTime = v
+	if ctx != nil {
+		t := ctx.Value("fromTime")
+		if t != nil {
+			if v, ok := t.(time.Time); ok {
+				fromTime = v
+			}
 		}
 	}
+
 	if err := pullK3CloudService.SyncDataProductPlan(transactionContext.(*pg.TransactionContext), fromTime); err != nil {
-		log.Logger.Error(err.Error())
+		log.Logger.Error(err.Error(), map[string]interface{}{"task": "定时同步车间计划"})
 		return nil
 	}
 	if err = transactionContext.CommitTransaction(); err != nil {
--- a/pkg/infrastructure/mqtt/subscribe_client.go
查看文件 @696a895
+++ b/pkg/infrastructure/mqtt/subscribe_client.go
查看文件 @696a895
@@ -3,8 +3,9 @@ package mqtt
 import (
 	"fmt"
 	pahomqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/linmadan/egglib-go/log"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
-	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
+	"runtime/debug"
 	"time"
 )
@@ -12,12 +13,15 @@ type SubscribeClient struct {
 	topic   string
 	handler pahomqtt.MessageHandler
 	client  pahomqtt.Client
+	log     log.Logger
 }
 type MessageHandler pahomqtt.MessageHandler
-func NewSubscribeClient() *SubscribeClient {
-	return &SubscribeClient{}
+func NewSubscribeClient(log log.Logger) *SubscribeClient {
+	return &SubscribeClient{
+		log: log,
+	}
 }
 func (subscribeClient *SubscribeClient) options() *pahomqtt.ClientOptions {
@@ -34,28 +38,32 @@ func (subscribeClient *SubscribeClient) options() *pahomqtt.ClientOptions {
 func (subscribeClient *SubscribeClient) Connect() *SubscribeClient {
 	opts := subscribeClient.options()
-	fmt.Println("start connect......")
+	subscribeClient.log.Info("mqtt start connect......")
 	opts.OnConnectionLost = func(c pahomqtt.Client, err error) {
-		defer func() {
-			if r := recover(); r != nil {
-				log.Logger.Error(fmt.Sprintf("%v", r))
-			}
-		}()
-		fmt.Println("Connect error:", err)
-		for {
-			fmt.Println("reconnect server")
-			token := subscribeClient.client.Connect()
-			token.Wait()
-			fmt.Println("server Connect status:", subscribeClient.client.IsConnectionOpen())
-			if subscribeClient.client.IsConnectionOpen() {
-				break
-			}
-			time.Sleep(3 * time.Second)
-		}
+		//defer func() {
+		//	if r := recover(); r != nil {
+		//		subscribeClient.log.Info(fmt.Sprintf("%v %s", r, debug.Stack()))
+		//	}
+		//}()
+		subscribeClient.log.Info("mqtt connect lost,error:" + err.Error())
+		//for {
+		//	subscribeClient.log.Info("reconnect server")
+		//	token := subscribeClient.client.Connect()
+		//	token.Wait()
+		//	subscribeClient.log.Info(fmt.Sprintf("server Connect status:%v", subscribeClient.client.IsConnectionOpen()))
+		//	if subscribeClient.client.IsConnectionOpen() {
+		//		break
+		//	}
+		//	time.Sleep(3 * time.Second)
+		//}
 	}
 	opts.OnConnect = func(c pahomqtt.Client) {
+		subscribeClient.log.Info("mqtt reconnected")
 		c.Subscribe(subscribeClient.topic, 0, subscribeClient.handler)
 	}
+	opts.OnReconnecting = func(client pahomqtt.Client, options *pahomqtt.ClientOptions) {
+		subscribeClient.log.Info("mqtt reconnecting...")
+	}
 	subscribeClient.client = pahomqtt.NewClient(opts)
 	token := subscribeClient.client.Connect()
 	token.Wait()
@@ -70,13 +78,13 @@ func (subscribeClient *SubscribeClient) Subscribe(topic string, messageHandler p
 	token.Done()
 }
-func StartSubscribe(topic string, handler MessageHandler) {
+func StartSubscribe(topic string, handler MessageHandler, log log.Logger) {
 	defer func() {
 		if err := recover(); err != nil {
-			fmt.Println(err)
-			StartSubscribe(topic, handler)
+			log.Error(fmt.Sprintf("%s", debug.Stack()))
+			StartSubscribe(topic, handler, log)
 		}
 	}()
-	fmt.Println("start subscribe...")
-	NewSubscribeClient().Connect().Subscribe(topic, pahomqtt.MessageHandler(handler))
+	log.Info("mqtt start subscribe...")
+	NewSubscribeClient(log).Connect().Subscribe(topic, pahomqtt.MessageHandler(handler))
 }
--- a/pkg/log/logrus.go
查看文件 @696a895
+++ b/pkg/log/logrus.go
查看文件 @696a895
 package log
 import (
+	"github.com/linmadan/egglib-go/log"
 	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/writer"
 	"io"
@@ -47,7 +48,7 @@ func (logger *LogrusLogger) AddHook(w io.Writer) {
 	level := logger.logrus.Level
 	var levels []logrus.Level
 	// 默认已经添加了一个当前log level的hook,所以此处 level+1
-	for i := 0; i < (int(level)); i++ {
+	for i := 0; i <= (int(level)); i++ {
 		levels = append(levels, logrus.Level(i))
 	}
 	logger.logrus.AddHook(&writer.Hook{
@@ -120,3 +121,5 @@ func NewLogrusLogger() *LogrusLogger {
 		logrus: logger,
 	}
 }
+
+var _ log.Logger = (*LogrusLogger)(nil)
--- a/pkg/port/mqtt/mqtt.go
查看文件 @696a895
+++ b/pkg/port/mqtt/mqtt.go
查看文件 @696a895
@@ -2,6 +2,7 @@ package mqtt
 import (
 	pahomqtt "github.com/eclipse/paho.mqtt.golang"
+	logimp "github.com/linmadan/egglib-go/log"
 	"github.com/linmadan/egglib-go/utils/json"
 	"github.com/tidwall/gjson"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/deviceCollection/command"
@@ -14,8 +15,8 @@ import (
 	"time"
 )
-func Start() {
-	mqtt.StartSubscribe(constant.MQTT_TOPIC, OnReceiveData)
+func Start(log logimp.Logger) {
+	mqtt.StartSubscribe(constant.MQTT_TOPIC, OnReceiveData, log)
 }
 func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {