修改代码

src/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs

@@ -132,8 +132,9 @@ namespace JiShe.CollectBus.ScheduledMeterReading
 
                 _logger.LogInformation($"{nameof(CreateToBeIssueTasks)} {timeDensity}分钟采集待下发任务创建完成");
 
-                //删除已经处理过的缓存数据
+                //根据当前的采集频率和类型，重新更新下一个任务点，把任务的创建源固定在当前逻辑，避免任务处理的逻辑异常导致任务创建失败。
-                await FreeRedisProvider.Instance.DelAsync(item);
+                tasksToBeIssueModel.NextTask = tasksToBeIssueModel.NextTask.AddMinutes(timeDensity);
+                await FreeRedisProvider.Instance.SetAsync(item, tasksToBeIssueModel);
             }
         }
 
@@ -326,9 +327,9 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList, currentTime);
             }
 
-            //删除任务数据
+            ////删除任务数据
-            await FreeRedisProvider.Instance.DelAsync(oneMinutekeyList);
+            //await FreeRedisProvider.Instance.DelAsync(oneMinutekeyList);
-            await CacheNextTaskData(timeDensity, MeterTypeEnum.Ammeter);
+            //await CacheNextTaskData(timeDensity, MeterTypeEnum.Ammeter);
 
 
             _logger.LogInformation($"{nameof(AmmeterScheduledMeterOneMinuteReading)} {timeDensity}分钟采集电表数据处理完成");
@@ -387,11 +388,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList, currentTime);
             }
 
-            //删除任务数据
+            ////删除任务数据
-            await FreeRedisProvider.Instance.DelAsync(fiveMinutekeyList);
+            //await FreeRedisProvider.Instance.DelAsync(fiveMinutekeyList);
 
-            //缓存下一个时间的任务
+            ////缓存下一个时间的任务
-            await CacheNextTaskData(timeDensity, MeterTypeEnum.Ammeter);
+            //await CacheNextTaskData(timeDensity, MeterTypeEnum.Ammeter);
 
             _logger.LogInformation($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} {timeDensity}分钟采集电表数据处理完成");
         }
@@ -452,11 +453,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList, currentDateTime);
             }
 
-            //删除任务数据
+            ////删除任务数据
             //await FreeRedisProvider.Instance.DelAsync(fifteenMinutekeyList);
 
-            //缓存下一个时间的任务
+            ////缓存下一个时间的任务
-            await CacheNextTaskData(timeDensity, MeterTypeEnum.Ammeter);
+            //await CacheNextTaskData(timeDensity, MeterTypeEnum.Ammeter);
 
             stopwatch.Stop();
 
@@ -703,6 +704,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                             AFN = aFN,
                             Fn = fn,
                             ItemCode = tempItem,
+                            TaskMark = CommonHelper.GetTaskMark((int)aFN, fn,ammeter.MeteringCode),
                             ManualOrNot = false,
                             Pn = ammeter.MeteringCode,
                             IssuedMessageId = GuidGenerator.Create().ToString(),
@@ -838,11 +840,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList);
             }
 
-            //删除任务数据
+            ////删除任务数据
-            await FreeRedisProvider.Instance.DelAsync(oneMinutekeyList);
+            //await FreeRedisProvider.Instance.DelAsync(oneMinutekeyList);
 
-            //缓存下一个时间的任务
+            ////缓存下一个时间的任务
-            await CacheNextTaskData(timeDensity, MeterTypeEnum.WaterMeter);
+            //await CacheNextTaskData(timeDensity, MeterTypeEnum.WaterMeter);
 
 
             _logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} {timeDensity}分钟采集水表数据处理完成");
@@ -900,11 +902,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList);
             }
 
-            //删除任务数据
+            ////删除任务数据
-            await FreeRedisProvider.Instance.DelAsync(fiveMinutekeyList);
+            //await FreeRedisProvider.Instance.DelAsync(fiveMinutekeyList);
 
-            //缓存下一个时间的任务
+            ////缓存下一个时间的任务
-            await CacheNextTaskData(timeDensity, MeterTypeEnum.WaterMeter);
+            //await CacheNextTaskData(timeDensity, MeterTypeEnum.WaterMeter);
 
 
             _logger.LogInformation($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} {timeDensity}分钟采集水表数据处理完成");
@@ -961,11 +963,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList);
             }
 
-            //删除任务数据
+            ////删除任务数据
-            await FreeRedisProvider.Instance.DelAsync(fifteenMinutekeyList);
+            //await FreeRedisProvider.Instance.DelAsync(fifteenMinutekeyList);
 
-            //缓存下一个时间的任务
+            ////缓存下一个时间的任务
-            await CacheNextTaskData(timeDensity, MeterTypeEnum.WaterMeter);
+            //await CacheNextTaskData(timeDensity, MeterTypeEnum.WaterMeter);
 
 
             _logger.LogInformation($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} {timeDensity}分钟采集水表数据处理完成");
@@ -975,7 +977,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
 
         #region 公共处理方法
         /// <summary>
-        /// 批量获取缓存的表计信息
+        /// Lua脚本批量获取缓存的表计信息
         /// </summary>
         /// <typeparam name="T">表信息数据对象</typeparam>
         /// <param name="redisKeys">采集频率对应的缓存Key集合</param>
@@ -1050,24 +1052,24 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             return true;
         }
 
-        /// <summary>
+        ///// <summary>
-        /// 缓存下一个时间的任务
+        ///// 缓存下一个时间的任务
-        /// </summary>
+        ///// </summary>
-        /// <param name="timeDensity">采集频率</param>
+        ///// <param name="timeDensity">采集频率</param>
-        /// <param name="meterType">表类型</param>
+        ///// <param name="meterType">表类型</param>
-        /// <returns></returns>
+        ///// <returns></returns>
-        private async Task CacheNextTaskData(int timeDensity, MeterTypeEnum meterType)
+        //private async Task CacheNextTaskData(int timeDensity, MeterTypeEnum meterType)
-        {
+        //{
-            //缓存下一个时间的任务
+        //    //缓存下一个时间的任务
-            TasksToBeIssueModel nextTask = new TasksToBeIssueModel()
+        //    TasksToBeIssueModel nextTask = new TasksToBeIssueModel()
-            {
+        //    {
-                TimeDensity = timeDensity,
+        //        TimeDensity = timeDensity,
-                NextTask = DateTime.Now.AddMinutes(timeDensity)
+        //        NextTask = DateTime.Now.AddMinutes(timeDensity)
-            };
+        //    };
 
-            var redisCacheKey = string.Format(RedisConst.CacheTasksToBeIssuedKey, SystemType, ServerTagName, meterType, timeDensity);
+        //    var redisCacheKey = string.Format(RedisConst.CacheTasksToBeIssuedKey, SystemType, ServerTagName, meterType, timeDensity);
-            await FreeRedisProvider.Instance.SetAsync(redisCacheKey, nextTask);
+        //    await FreeRedisProvider.Instance.SetAsync(redisCacheKey, nextTask);
-        }
+        //}
 
 
         /// <summary>

src/JiShe.CollectBus.Common/Helpers/CommonHelper.cs

@@ -760,5 +760,17 @@ namespace JiShe.CollectBus.Common.Helpers
 
             return fontValue;
         } 
+
+        /// <summary>
+        /// 获取任务标识
+        /// </summary>
+        /// <param name="afn"></param>
+        /// <param name="fn"></param>
+        /// <param name="pn"></param>
+        /// <returns></returns>
+        public static string GetTaskMark(int afn,int fn,int pn)
+        {
+            return $"{afn.ToString().PadLeft(2,'0')}{fn}{pn}";
+        }
     }
 }

src/JiShe.CollectBus.Domain/IotSystems/Devices/Device.cs

@@ -57,6 +57,11 @@ namespace JiShe.CollectBus.IotSystems.Devices
         /// </summary>
         public DeviceStatus Status { get; set; }
 
+        /// <summary>
+        /// 设备任务超时次数，超过一定次数则发出预警。
+        /// </summary>
+        public int TaskTimeOutCounts { get; set; } = 0;
+
         public void UpdateByLoginAndHeartbeat(string clientId)
         {
             LastOnlineTime = DateTime.Now;

src/JiShe.CollectBus.Domain/IotSystems/MeterReadingRecords/MeterReadingRecords.cs

@@ -10,7 +10,7 @@ using Volo.Abp.Domain.Entities.Auditing;
 namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
 {
     /// <summary>
-    /// 抄读数据记录
+    /// 抄读任务数据记录
     /// </summary>
     public class MeterReadingRecords : AggregateRoot<Guid>
     {  
@@ -19,21 +19,26 @@ namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
         /// </summary>
         public bool ManualOrNot { get; set; }
 
+        /// <summary>
+        /// 任务数据唯一标记
+        /// </summary>
+        public string TaskMark { get; set; }
+
+        /// <summary>
+        /// 时间戳标记，IoTDB时间列处理，上报通过构建标记获取唯一标记匹配时间戳。
+        /// </summary>
+        public long Timestamps { get; set; }
+
+        /// <summary>
+        /// 是否超时
+        /// </summary>
+        public bool IsTimeout { get; set; } = false;
+
         /// <summary>
         /// 待抄读时间
         /// </summary>
         public DateTime PendingCopyReadTime { get; set; }
          
-        /// <summary>
-        /// 下发消息内容
-        /// </summary>
-        public string IssuedMessageHexString { get; set; }
-
-        /// <summary>
-        /// 下发消息Id
-        /// </summary>
-        public string IssuedMessageId { get; set; }
-         
         /// <summary>
         /// 集中器ID
         /// </summary>
@@ -89,16 +94,22 @@ namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
         /// </summary>
         public string ItemCode { get; set;}
          
-        /// <summary>
-        /// 是否超时
-        /// </summary>
-        public bool IsTimeout { get; set; } = false;
 
         /// <summary>
         /// 创建时间
         /// </summary>
         public DateTime CreationTime { get; set; }
 
+        /// <summary>
+        /// 下发消息内容
+        /// </summary>
+        public string IssuedMessageHexString { get; set; }
+
+        /// <summary>
+        /// 下发消息Id
+        /// </summary>
+        public string IssuedMessageId { get; set; }
+
         /// <summary>
         /// 消息上报内容
         /// </summary>