Merge remote-tracking branch 'origin/test'

@@ -117,7 +117,7 @@ spec:
         - name: MANUFACTURE_DEFAULT_ORGID
           value: "44"
         - name: MANUFACTURE_DEFAULT_WORKSHOPID
-          value: "1"
+          value: "2"
         - name: MANUFACTURE_PRODUCT_TYPE
           value: "0502010001ST,0502010007ST,0502010008ST,0502010009ST,0502010010ST,0502010011ST,0502010014ST,0502010016ST,0502010018ST,0502010020ST,0502010021ST,0502010022ST,0502010023ST"
         - name: MQTT_HOST

@@ -122,8 +122,9 @@ func (deviceCollectionService *DeviceCollectionService) DeviceCollection(createD
 	}
 
 	//计算区间的产能
+
 	if v, ok := newDeviceCollection.Values["Count"]; ok {
-		newDeviceCollection.Values["total"] = v // 记录原始值
+		newDeviceCollection.Values["Total"] = v // 记录原始值
 		newDeviceCollection.Values["Count"] = 0
 		curCount, errCurCount := strconv.Atoi(utils.AssertString(v))
 		lastCount, errLastCount := strconv.Atoi(utils.AssertString(lastDeviceCollectionRecord.Values["Count"]))
@@ -142,26 +143,27 @@ func (deviceCollectionService *DeviceCollectionService) DeviceCollection(createD
 			newDeviceCollection.ProductCount = count
 		}
 	}
+	if _, valErr := newDeviceCollection.Valid(); valErr != nil {
+		newDeviceCollection.ResetProductCountToZero()
+	}
 	// TODO:测试假数据，后期注释掉
 	//if createDeviceCollectionCommand.DeviceType == domain.DeviceTypeChuanChuanJi {
 	//	newDeviceCollection.Values["Count"] = rand.Intn(300)
+	//	newDeviceCollection.StartupStatus = 1
+	//	newDeviceCollection.ComStatus = 1
+	//	newDeviceCollection.CollectionTime = time.Date(2022,4,18,7,0,0,0,time.Local)
 	//}
-	deviceCollection, err := deviceCollectionRepository.Save(newDeviceCollection)
-	if err != nil {
+	if newDeviceCollection, err = deviceCollectionRepository.Save(newDeviceCollection); err != nil {
 		return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
 	}
-	err = domainService.SendWorkshopDeviceData(deviceCollection)
-	if err != nil {
+	if err = domainService.SendWorkshopDeviceData(newDeviceCollection); err != nil {
 		log.Logger.Error("车间设备数据加入redis失败：" + err.Error())
 		return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
 	}
-
 	if err := transactionContext.CommitTransaction(); err != nil {
 		return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error())
 	}
-	return map[string]interface{}{
-		"deviceCollection": deviceCollection,
-	}, nil
+	return nil, nil
 
 }
 

@@ -2,6 +2,7 @@ package dto
 
 import (
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/utils"
 	"time"
 )
 
@@ -43,7 +44,7 @@ func (d *ProductLevelTwoRecord) LoadDto(m *domain.ProductRecord, orgId int) *Pro
 	d.ProductRecordId = m.ProductRecordId
 	d.ProductWorker = m.ProductWorker
 	d.WorkStation = m.WorkStation
-	d.WeighBefore = m.ProductRecordInfo.WeighBefore
+	d.WeighBefore = utils.Round(m.ProductRecordInfo.WeighBefore, 1)
 	d.WeighAfter = m.ProductRecordInfo.WeighAfter
 	d.ApproveStatus = m.ProductRecordInfo.ApproveStatus
 	d.WorkOn = m.ProductRecordInfo.WorkOn

@@ -291,12 +291,6 @@ func (productRecordService *ProductRecordService) ProductRecordStatics(cmd *comm
 	}()
 	var _ domain.ProductRecordRepository
 	var productRecord *domain.ProductRecord = cmd.ProductRecord
-	//_,productRecord,err = factory.FastPgProductRecord(transactionContext,cmd.ProductRecordId)
-	//if err!=nil{
-	//	log.Logger.Error(err.Error())
-	//	return nil, nil
-	//}
-	//
 	if productRecord == nil {
 		return nil, nil
 	}

@@ -97,3 +97,26 @@ func (deviceCollection *DeviceCollection) Update(data map[string]interface{}) er
 func TaskDeviceCollection() string {
 	return fmt.Sprintf("%v:task:device-collection:report", constant.CACHE_PREFIX)
 }
+
+func (data *DeviceCollection) Valid() (bool, error) {
+	var (
+		result = false
+	)
+	if data.ProductCount > DeviceMaxSingleProductCount {
+		return result, fmt.Errorf("设备数据异常: 生产数量超过:%v", DeviceMaxSingleProductCount)
+	}
+	if data.StartupStatus == 0 {
+		return result, fmt.Errorf("设备数据异常: 启动0")
+	}
+	if data.ComStatus == 0 {
+		return result, fmt.Errorf("设备数据异常: 通讯0")
+	}
+	return true, nil
+}
+
+func (data *DeviceCollection) ResetProductCountToZero() {
+	data.ProductCount = 0
+	if data.Values != nil {
+		data.Values["Count"] = 0
+	}
+}

@@ -53,14 +53,19 @@ func (deviceDailyRunningRecord *DeviceDailyRunningRecord) Update(data map[string
 	return nil
 }
 
-func (deviceDailyRunningRecord *DeviceDailyRunningRecord) AddDeviceRunningData(t time.Time, data *DeviceRunningData) {
+func (deviceDailyRunningRecord *DeviceDailyRunningRecord) AddDeviceRunningData(data *DeviceRunningData) bool {
+	t := data.CollectionTime
 	deviceDailyRunningRecord.DeviceRunningRecordInfo.AddDeviceRunningData(t, data)
 	now := time.Now().Unix()
-	if t.Unix() > (now-DefaultCollectionTimeSpan) && t.Unix() < (now+DefaultCollectionTimeSpan) {
+	ts := t.Local().Unix()
+	deviceDailyRunningRecord.UpdatedAt = time.Now()
+	if ts > (now-DefaultCollectionTimeSpan) && ts < (now+DefaultCollectionTimeSpan) {
 		deviceDailyRunningRecord.UpdatedAt = t
-		return
 	}
-	deviceDailyRunningRecord.UpdatedAt = time.Now()
+	if ok, _ := data.Valid(); !ok {
+		return false
+	}
+	return true
 }
 
 func (deviceDailyRunningRecord *DeviceDailyRunningRecord) String() string {

 package domain
 
-import "time"
+import (
+	"fmt"
+	"time"
+)
+
+const DeviceMaxSingleProductCount = 10000
 
 // 设备运行数据
 type DeviceRunningData struct {
@@ -24,6 +29,8 @@ type DeviceRunningData struct {
 	// 附加数据
 	// 匹配数目
 	Count int `json:"count"`
+	// 当天数量
+	TodayTotal int `json:"today_total"`
 	// 合计数目
 	Total int `json:"total"`
 	// 炸机前段温度：炸机前段当前温度 YZJ1 油炸机
@@ -41,3 +48,20 @@ type DeviceRunningData struct {
 	// 单位数据 比如：1串/0.1kg   weight = count * unitQuantity
 	UnitQuantity float64 `json:"unitQuantity"`
 }
+
+func (data *DeviceRunningData) Valid() (bool, error) {
+	var (
+		result = false
+	)
+	if data.Count > DeviceMaxSingleProductCount {
+		data.Count = 0
+		return result, fmt.Errorf("设备数据异常: 生产数量超过:%v", DeviceMaxSingleProductCount)
+	}
+	if data.StartupStatus == 0 {
+		return result, fmt.Errorf("设备数据异常: 启动0")
+	}
+	if data.ComStatus == 0 {
+		return result, fmt.Errorf("设备数据异常: 通讯0")
+	}
+	return true, nil
+}

@@ -75,7 +75,7 @@ func (info *ProductRecordStaticInfo) PreStatistics(productWeight float64, second
 	} else if info.OutputWeight == info.InputWeight {
 		info.QualificationRate = 100
 	} else {
-		info.QualificationRate = utils.Round(info.OutputWeight*100.0/info.InputWeight, 0)
+		info.QualificationRate = utils.Truncate(info.OutputWeight*100.0/info.InputWeight, 2)
 	}
 }
 

 package domainService
 
 import (
+	"fmt"
 	"github.com/hibiken/asynq"
 	"github.com/linmadan/egglib-go/utils/json"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
 )
 
-func SendWorkshopWorkTimeStaticJob(productRecord *domain.ProductAttendanceRecord) error {
-	return SendAsyncJob(domain.TaskKeyWorkshopWorkTimeRecordStatics(), productRecord)
+const (
+	QueueProduct = "product"
+	QueueDevice  = "device"
+	QueueDefault = "default"
+)
+
+func FormatQueue(qt string) string {
+	return fmt.Sprintf("%v:queue:%v", constant.CACHE_PREFIX, qt)
 }
 
-func SendProductRecordStaticsJob(productRecord *domain.ProductRecord) error {
-	task := asynq.NewTask(domain.TaskKeyPatternProductRecordStatics(), []byte(json.MarshalToString(productRecord)))
+func SendWorkshopWorkTimeStaticJob(r *domain.ProductAttendanceRecord) error {
+	return SendAsyncJob(domain.TaskKeyWorkshopWorkTimeRecordStatics(), r, asynq.Queue(FormatQueue(QueueDefault)))
+}
 
-	client := asynq.NewClient(asynq.RedisClientOpt{Addr: constant.REDIS_ADDRESS})
-	_, err := client.Enqueue(task)
-	return err
+func SendProductRecordStaticsJob(r *domain.ProductRecord) error {
+	return SendAsyncJob(domain.TaskKeyPatternProductRecordStatics(), r, asynq.Queue(FormatQueue(QueueProduct)))
 }
 
-func SendDeviceZkTecoReportJob(productRecord *domain.DeviceZkTeco) error {
-	return SendAsyncJob(domain.TaskDeviceZkTecoReport(), productRecord)
+func SendDeviceZkTecoReportJob(r *domain.DeviceZkTeco) error {
+	return SendAsyncJob(domain.TaskDeviceZkTecoReport(), r, asynq.Queue(FormatQueue(QueueDefault)))
 }
 
-func SendWorkshopDeviceData(productRecord *domain.DeviceCollection) error {
-	return SendAsyncJob(domain.TaskDeviceCollection(), productRecord)
+func SendWorkshopDeviceData(r *domain.DeviceCollection) error {
+	return SendAsyncJob(domain.TaskDeviceCollection(), r, asynq.Queue(FormatQueue(QueueDevice)))
 }
 
-func SendAsyncJob(queueName string, job interface{}) error {
+func SendAsyncJob(queueName string, job interface{}, opts ...asynq.Option) error {
 	task := asynq.NewTask(queueName, []byte(json.MarshalToString(job)))
 	client := asynq.NewClient(asynq.RedisClientOpt{Addr: constant.REDIS_ADDRESS})
-	_, err := client.Enqueue(task)
+	_, err := client.Enqueue(task, opts...)
 	return err
 }

@@ -60,7 +60,7 @@ func (ptr *PGProductRecordService) SubmitProductRecord(productRecordType int, qu
 		workstation *domain.WorkStation
 		uc          *domain.UnitConversion
 		err         error
-		weight      float64 = request.Weigh
+		weight      float64 = utils.Round(request.Weigh, 1)
 	)
 	if plan, err = productPlanRepository.FindOne(map[string]interface{}{"productPlanId": request.ProductPlanId}); err != nil {
 		return nil, err

@@ -25,10 +25,8 @@ const (
 func (ptr *PGProductRecordService) EmployeeProductStatics(productRecord *domain.ProductRecord) (interface{}, error) {
 
 	var (
-		workshopRepository, _     = repository.NewWorkshopRepository(ptr.transactionContext)
-		productPlanRepository, _  = repository.NewProductPlanRepository(ptr.transactionContext)
-		productGroupRepository, _ = repository.NewProductGroupRepository(ptr.transactionContext)
-		//employeeProductRecordRepository, _ = repository.NewEmployeeProductRecordRepository(ptr.transactionContext)
+		workshopRepository, _    = repository.NewWorkshopRepository(ptr.transactionContext)
+		productPlanRepository, _ = repository.NewProductPlanRepository(ptr.transactionContext)
 	)
 
 	var (
@@ -60,9 +58,9 @@ func (ptr *PGProductRecordService) EmployeeProductStatics(productRecord *domain.
 	case ProductSection3:
 		break
 	case ProductSection4: //个人特殊处理
-		return ptr.personalProductStatics(nil, nil, nil, productRecord)
+		return ptr.personalProductStatics(nil, productRecord)
 	default:
-		return nil, nil //ptr.personalProductStatics(productRecord)
+		return nil, nil
 	}
 	if planId == 0 {
 		log.Logger.Debug(fmt.Sprintf("工段:%v product_record 编号:%v 批次为0", productRecord.WorkStation.WorkStationId, productRecord.ProductRecordId))
@@ -74,9 +72,6 @@ func (ptr *PGProductRecordService) EmployeeProductStatics(productRecord *domain.
 		return nil, err
 	}
 
-	// 2.1 判断是否是支援类型  有打卡记录,员工是否是属于该工段的员工
-	groupMembers, groupMembersKeyFunc := FindGroupMembers(productGroupRepository, cid, oid, productRecord.WorkStation.WorkStationId)
-
 	// 集体
 	// 1.查询员工 -》 员工打卡记录  工位+打卡日期
 	// 2.打卡记录的时间区间 在生产记录上报的时间范围内
@@ -91,8 +86,8 @@ func (ptr *PGProductRecordService) EmployeeProductStatics(productRecord *domain.
 		productRecord.ProductWorker = r.ProductWorker
 		// 3.查询员工产能记录 -》员工 批次+工位+批次生产日期 （有 更新产能数据、没有插入一条产能数据）
 		// 4.更新产能 (产能、二级品) （特殊工段处理 打料、成型）
-		// 个人
-		if _, err := ptr.personalProductStatics(productPlan, groupMembers, groupMembersKeyFunc, productRecord); err != nil {
+		// 个人产能统计
+		if _, err := ptr.personalProductStatics(productPlan, productRecord); err != nil {
 			return nil, err
 		}
 	}
@@ -174,31 +169,28 @@ func FindGroupMembers(productGroupRepository domain.ProductGroupRepository, comp
 }
 
 // 个人生产记录统计
-func (ptr *PGProductRecordService) personalProductStatics(productPlan *domain.ProductPlan, groupMembers map[string]*domain.User, groupMembersKeyFunc func(int) string, productRecord *domain.ProductRecord) (interface{}, error) {
+func (ptr *PGProductRecordService) personalProductStatics(productPlan *domain.ProductPlan, productRecord *domain.ProductRecord) (interface{}, error) {
 	var (
-		//workshopRepository,_=repository.NewWorkshopRepository(ptr.transactionContext)
 		productPlanRepository, _           = repository.NewProductPlanRepository(ptr.transactionContext)
 		productGroupRepository, _          = repository.NewProductGroupRepository(ptr.transactionContext)
 		employeeProductRecordRepository, _ = repository.NewEmployeeProductRecordRepository(ptr.transactionContext)
 	)
 	var (
-		cid    = productRecord.CompanyId
-		oid    = productRecord.OrgId
-		planId = productRecord.ProductRecordInfo.ProductPlanId
-		//productPlan *domain.ProductPlan
-		err error
+		cid           = productRecord.CompanyId
+		oid           = productRecord.OrgId
+		planId        = productRecord.ProductRecordInfo.ProductPlanId
+		err           error
+		workStationId = productRecord.WorkStation.WorkStationId
 	)
 	// 2.1 判断是否是支援类型  有打卡记录,员工是否是属于该工段的员工
-	if groupMembers == nil {
-		groupMembers, groupMembersKeyFunc = FindGroupMembers(productGroupRepository, cid, oid, productRecord.WorkStation.WorkStationId)
-	}
+	groupMembers, groupMembersKeyFunc := FindGroupMembers(productGroupRepository, cid, oid, workStationId)
+
 	if productPlan == nil {
 		productPlan, err = productPlanRepository.FindOne(map[string]interface{}{"productPlanId": planId})
 		if err != nil {
 			return nil, err
 		}
 	}
-	workStationId := productRecord.WorkStation.WorkStationId
 
 	employeeProductRecordDao, _ := dao.NewEmployeeProductRecordDao(ptr.transactionContext)
 
@@ -231,11 +223,13 @@ func (ptr *PGProductRecordService) personalProductStatics(productPlan *domain.Pr
 			}
 		}
 	}
+	log.Logger.Debug(fmt.Sprintf("产能统计 员工:%v(%v) 当前产能:%v kg 新增产能:%v kg  原始数量:%v",
+		employeeProductRecord.ProductWorker.UserName, employeeProductRecord.ProductWorker.UserId,
+		employeeProductRecord.ProductWeigh, productRecord.ProductRecordInfo.Weigh, productRecord.ProductRecordInfo.Original))
 
 	employeeProductRecord.UpdateProductWeigh(productRecord, yesterdayOutputWeight, bestOutputWeight)
 
 	if employeeProductRecord, err = employeeProductRecordRepository.Save(employeeProductRecord); err != nil {
-		// TODO：异常处理
 		log.Logger.Error(fmt.Sprintf("生产记录:[%v] 员工:[%v] 处理异常:%v", productRecord.ProductRecordId, productRecord.ProductWorker.UserId, err.Error()))
 	}
 	return nil, nil
@@ -244,9 +238,8 @@ func (ptr *PGProductRecordService) personalProductStatics(productPlan *domain.Pr
 //WorkshopProductStatics  车间产能统计
 func (ptr *PGProductRecordService) WorkshopProductStatics(productRecord *domain.ProductRecord) (interface{}, error) {
 	var (
-		workshopRepository, _    = repository.NewWorkshopRepository(ptr.transactionContext)
-		productPlanRepository, _ = repository.NewProductPlanRepository(ptr.transactionContext)
-		//productGroupRepository, _          = repository.NewProductGroupRepository(ptr.transactionContext)
+		workshopRepository, _              = repository.NewWorkshopRepository(ptr.transactionContext)
+		productPlanRepository, _           = repository.NewProductPlanRepository(ptr.transactionContext)
 		workshopProductRecordRepository, _ = repository.NewWorkshopProductRecordRepository(ptr.transactionContext)
 	)
 
@@ -295,6 +288,9 @@ func (ptr *PGProductRecordService) WorkshopProductStatics(productRecord *domain.
 			return nil, nil
 		}
 	}
+	log.Logger.Debug(fmt.Sprintf("产能统计 工位:%v(%v) 当前产能:%v kg 新增产能:%v kg  原始数量:%v",
+		workshopProductRecord.WorkStation.WorkStationId, workshopProductRecord.WorkStation.SectionName,
+		workshopProductRecord.ProductWeigh, productRecord.ProductRecordInfo.Weigh, productRecord.ProductRecordInfo.Original))
 	workshopProductRecord.UpdateProductWeigh(productRecord)
 	// 打料 跟 成型工段的初始产能是批次的产能
 	if productRecord.WorkStation.SectionName == ProductSection1 && productRecord.WorkStation.SectionName == ProductSection2 {

@@ -28,7 +28,6 @@ func (ptr *PGWorkshopDataConsumeService) Consume(companyId, orgId int, record *d
 		planId              int
 		err                 error
 		plan                *domain.ProductPlanDispatchRecord
-		//datetime            time.Time
 	)
 	var (
 		deviceRepository, _              = repository.NewDeviceRepository(ptr.transactionContext)
@@ -67,12 +66,11 @@ func (ptr *PGWorkshopDataConsumeService) Consume(companyId, orgId int, record *d
 
 		var saveErr error
 		if deviceDailyRecord, saveErr = ptr.saveDeviceDailyRunningRecord(companyId, orgId, workStation, device, planId, deviceRunningData); err != nil {
-			//log.Logger.Error(saveErr.Error())
 			return nil, err
 		}
 		defer func() {
 			if saveErr != nil {
-				redis.RemoveDeviceDailyRunningRecord(time.Now(), deviceRunningData.DeviceCode)
+				redis.RemoveDeviceDailyRunningRecord(deviceRunningData.CollectionTime, deviceRunningData.DeviceCode)
 			}
 		}()
 	}
@@ -88,24 +86,31 @@ func (ptr *PGWorkshopDataConsumeService) Consume(companyId, orgId int, record *d
 	}
 
 	// 1.保存设备运行记录
+	deviceRunningData.TodayTotal = deviceDailyRecord.DeviceRunningRecordInfo.Count
 	deviceRunningRecord, _ = ptr.newDeviceRunningRecord(companyId, orgId, workStation, device, deviceRunningData)
 	if _, err = deviceRunningRecordRepository.Save(deviceRunningRecord); err != nil {
 		return nil, err
 	}
-	// 2.保存设备生产记录 (统计车间、员工产能) 批次跟数量不为空
-	if record.DeviceType == domain.DeviceTypeChuanChuanJi && plan != nil && deviceRunningData.Count > 0 {
 
-		productRecord, _ := ptr.newProductRecord(companyId, orgId, workStation, device, deviceRunningData, plan)
-		//if _, err = deviceRunningRecordRepository.Save(deviceRunningRecord); err != nil {
-		//	return nil, err
-		//}
-		SendProductRecordStaticsJob(productRecord)
+	// 2.更新 设备每日运行记录（汇总） - redis更新 十分钟异步刷库 TODO:这边会有并发问题，加锁
+	if addSuccess := deviceDailyRecord.AddDeviceRunningData(deviceRunningData); addSuccess {
+		// 3.保存设备生产记录 (统计车间、员工产能) 批次跟数量不为空
+		if record.DeviceType == domain.DeviceTypeChuanChuanJi && plan != nil && deviceRunningData.Count > 0 {
+			log.Logger.Debug(fmt.Sprintf("设备统计 设备:%v(%v) 当前数量:%v 增加数量:%v",
+				deviceDailyRecord.DeviceCode,
+				deviceDailyRecord.ProductDate.Local().Format("2006-01-02"),
+				deviceRunningRecord.DeviceRunningRecordInfo.TodayTotal,
+				deviceRunningRecord.DeviceRunningRecordInfo.Count))
+			productRecord, _ := ptr.newProductRecord(companyId, orgId, workStation, device, deviceRunningData, plan)
+			// 同步执行
+			//productRecordService, _ := NewPGProductRecordService(ptr.transactionContext)
+			//productRecordService.EmployeeProductStatics(productRecord)
+			//productRecordService.WorkshopProductStatics(productRecord)
+			// 异步执行
+			SendProductRecordStaticsJob(productRecord)
+		}
 	}
-
-	// 3.更新 设备每日运行记录（汇总） - redis更新 十分钟异步刷库
-	deviceDailyRecord.AddDeviceRunningData(deviceRunningData.CollectionTime, deviceRunningData)
 	if err = redis.SaveDeviceDailyRunningRecord(deviceDailyRecord); err != nil {
-		//log.Logger.Error(err.Error())
 		return nil, err
 	}
 	return nil, nil
@@ -153,16 +158,16 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev
 			if err != nil {
 				break
 			}
-			data.Temp1 = deviceYouZhaJi.FrontTemp
-			data.Temp2 = deviceYouZhaJi.BackTemp
+			data.Temp1 = utils.Truncate(deviceYouZhaJi.FrontTemp, 1)
+			data.Temp2 = utils.Truncate(deviceYouZhaJi.BackTemp, 1)
 		} else {
 			deviceYouZhaJi := &domain.DeviceYouZhaJi2{}
 			err = json.Unmarshal(mBytes, deviceYouZhaJi)
 			if err != nil {
 				break
 			}
-			data.Temp1 = deviceYouZhaJi.Temp1
-			data.Temp2 = deviceYouZhaJi.Temp2
+			data.Temp1 = utils.Truncate(deviceYouZhaJi.Temp1, 1)
+			data.Temp2 = utils.Truncate(deviceYouZhaJi.Temp2, 1)
 		}
 		break
 	//串串机
@@ -187,7 +192,7 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev
 		if err != nil {
 			break
 		}
-		data.Temp1 = deviceSuDongXian.CurrTemp
+		data.Temp1 = utils.Truncate(deviceSuDongXian.CurrTemp, 1)
 		break
 	//封口机
 	case domain.DeviceTypeFengKouJi:
@@ -296,7 +301,7 @@ func (ptr *PGWorkshopDataConsumeService) saveDeviceDailyRunningRecord(companyId,
 		DeviceCode:              device.DeviceCode,
 		ProductDate:             utils.GetZeroTime(data.CollectionTime),
 		DeviceRunningRecordInfo: recordInfo,
-		CreatedAt:               time.Now(),
+		CreatedAt:               data.CollectionTime,
 		UpdatedAt:               time.Now(),
 	}
 	if record, err = deviceDailyRunningRecordRepository.Save(record); err != nil {

@@ -6,6 +6,7 @@ import (
 	"github.com/go-redis/redis"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
 	"time"
 )
 
@@ -13,6 +14,7 @@ import (
 func GetDeviceDailyRunningRecord(t time.Time, deviceCode string) (*domain.DeviceDailyRunningRecord, error) {
 	client := GetRedis()
 	key := DeviceDailyRunningRecordKey(t, deviceCode)
+	log.Logger.Debug(fmt.Sprintf("Redis  Device:%v GET Key:%v", deviceCode, key))
 	return getDeviceDailyRunningRecord(client, key)
 }
 
@@ -32,7 +34,8 @@ func getDeviceDailyRunningRecord(client *redis.Client, key string) (*domain.Devi
 // 保存每日设备运行数据
 func SaveDeviceDailyRunningRecord(record *domain.DeviceDailyRunningRecord) error {
 	client := GetRedis()
-	key := DeviceDailyRunningRecordKey(record.ProductDate.Local(), record.DeviceCode)
+	key := DeviceDailyRunningRecordKey(record.ProductDate, record.DeviceCode)
+	log.Logger.Debug(fmt.Sprintf("Redis  Device:%v SET Key:%v Count:%v", record.DeviceCode, key, record.DeviceRunningRecordInfo.Count))
 	recordData, err := json.Marshal(record)
 	result := client.Set(key, recordData, time.Hour*24*5)
 	_, err = result.Result()
@@ -42,7 +45,7 @@ func SaveDeviceDailyRunningRecord(record *domain.DeviceDailyRunningRecord) error
 // 保存每日设备运行数据
 func RemoveDeviceDailyRunningRecord(t time.Time, deviceCode string) error {
 	client := GetRedis()
-	key := DeviceDailyRunningRecordKey(t.Local(), deviceCode)
+	key := DeviceDailyRunningRecordKey(t, deviceCode)
 	result := client.Del(key)
 	_, err := result.Result()
 	return err

@@ -408,9 +408,17 @@ func NewSnowflakeId() (int64, error) {
 	return id.Int64(), nil
 }
 
+// Round 保留数值的精度位 四舍五入
 func Round(value float64, places int32) float64 {
 	quantity := decimal.NewFromFloat(value)
 	d := quantity.Round(places)
 	rsp, _ := d.Float64()
 	return rsp
 }
+
+// Truncate 截取数值固定长度的 eg:Truncate(99.99,1)  Result: 99.9
+func Truncate(value float64, places int32) float64 {
+	quantity := decimal.NewFromFloat(value).Truncate(places)
+	rsp, _ := quantity.Float64()
+	return rsp
+}

@@ -3,6 +3,7 @@ package utils
 import (
 	"fmt"
 	"github.com/stretchr/testify/assert"
+	"math"
 	"testing"
 	"time"
 )
@@ -67,6 +68,7 @@ func timeParse() {
 	fmt.Println("1. str: ", str)
 	t, _ := time.Parse(TIME_LAYOUT, str)
 	fmt.Println("2. Parse time: ", t)
+	fmt.Println("2.1. Parse time: ", t.Local(), t.Local().Local())
 	tStr := t.Format(TIME_LAYOUT)
 	fmt.Println("3. Format time str: ", tStr)
 	name, offset := t.Zone()
@@ -102,3 +104,14 @@ func parseWithLocation(name string, timeStr string) (time.Time, error) {
 		return lt, nil
 	}
 }
+
+func TestRound(t *testing.T) {
+	t.Logf("%v", Round(99.999, 1))
+	t.Logf("%v", Round(99.999, 2))
+
+	t.Logf("%.1f", math.Floor(99.99))
+	t.Logf("%v", math.Ceil(99.99))
+
+	t.Logf("%.1f", Truncate(99.99, 1))
+	t.Logf("%v", Truncate(99, 0))
+}

 package mqtt
 
 import (
+	"fmt"
 	pahomqtt "github.com/eclipse/paho.mqtt.golang"
 	logimp "github.com/linmadan/egglib-go/log"
 	"github.com/linmadan/egglib-go/utils/json"
@@ -32,6 +33,7 @@ func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
 			collectionTime, _ = time.ParseInLocation("2006-01-02 - 15:04:05", t.(string), time.Local)
 		}
 		if collectionTime.IsZero() {
+			log.Logger.Error(fmt.Sprintf("采集时间有误:%v ", collectionTime), map[string]interface{}{"data": message})
 			return
 		}
 		var mBytes []byte

@@ -5,6 +5,7 @@ import (
 	"github.com/hibiken/asynq"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/domainService"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
 	"os"
 	"os/signal"
@@ -19,7 +20,17 @@ func Run() {
 	}()
 	srv := asynq.NewServer(
 		asynq.RedisClientOpt{Addr: constant.REDIS_ADDRESS},
-		asynq.Config{Concurrency: 1},
+		asynq.Config{
+			//Concurrency: 4,
+			Queues: map[string]int{
+				//"critical": 1,
+				"default": 1,
+				domainService.FormatQueue(domainService.QueueDevice):  1,
+				domainService.FormatQueue(domainService.QueueProduct): 1,
+				domainService.FormatQueue(domainService.QueueDefault): 1,
+			},
+			StrictPriority: true,
+		},
 	)
 
 	h := asynq.NewServeMux()

@@ -17,8 +17,8 @@ func HandlerProductRecordStatics(c context.Context, t *asynq.Task) error {
 	if err := json.Unmarshal(t.Payload(), cmd); err != nil {
 		return err
 	}
-	log.Logger.Debug(fmt.Sprintf("【生产记录统计】 消费 生产记录ID：%v 类型:%v 工段:%v(%v) 重量:%v", cmd.ProductRecordId, cmd.ProductRecordType,
-		cmd.WorkStation.SectionName, cmd.WorkStation.WorkStationId, cmd.ProductRecordInfo.Weigh))
+	log.Logger.Debug(fmt.Sprintf("【生产记录统计】 消费 生产记录ID：%v 类型:%v 工段:%v(%v) 重量:%v  记录时间:%v", cmd.ProductRecordId, cmd.ProductRecordType,
+		cmd.WorkStation.SectionName, cmd.WorkStation.WorkStationId, cmd.ProductRecordInfo.Weigh, cmd.CreatedAt))
 	_, err := productPlanService.ProductRecordStatics(cmd)
 	if err != nil {
 		log.Logger.Error(err.Error())