提交代码

modules/JiShe.CollectBus.Kafka/AdminClient/AdminClientService.cs

@@ -1,7 +1,9 @@
 using Confluent.Kafka;
 using Confluent.Kafka.Admin;
+using JiShe.CollectBus.Kafka.Internal;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 using Volo.Abp.DependencyInjection;
 
 namespace JiShe.CollectBus.Kafka.AdminClient;
@@ -9,16 +11,17 @@ namespace JiShe.CollectBus.Kafka.AdminClient;
 public class AdminClientService : IAdminClientService, IDisposable, ISingletonDependency
 {
     private readonly ILogger<AdminClientService> _logger;
-
+    private readonly KafkaOptionConfig _kafkaOptionConfig;
     /// <summary>
     /// Initializes a new instance of the <see cref="AdminClientService" /> class.
     /// </summary>
     /// <param name="configuration"></param>
     /// <param name="logger"></param>
-    public AdminClientService(IConfiguration configuration, ILogger<AdminClientService> logger)
+    public AdminClientService(IConfiguration configuration, ILogger<AdminClientService> logger, IOptions<KafkaOptionConfig> kafkaOptionConfig)
     {
         _logger = logger;
-        Instance = GetInstance(configuration);
+        _kafkaOptionConfig = kafkaOptionConfig.Value;
+        Instance = GetInstance();
     }
 
     /// <summary>
@@ -142,22 +145,19 @@ public class AdminClientService : IAdminClientService, IDisposable, ISingletonDe
     /// <summary>
     ///     Gets the instance.
     /// </summary>
-    /// <param name="configuration">The configuration.</param>
     /// <returns></returns>
-    public IAdminClient GetInstance(IConfiguration configuration)
+    public IAdminClient GetInstance()
     {
-        ArgumentException.ThrowIfNullOrWhiteSpace(configuration["Kafka:EnableAuthorization"]);
-        var enableAuthorization = bool.Parse(configuration["Kafka:EnableAuthorization"]!);
         var adminClientConfig = new AdminClientConfig
         {
-            BootstrapServers = configuration["Kafka:BootstrapServers"]
+            BootstrapServers = _kafkaOptionConfig.BootstrapServers
         };
-        if (enableAuthorization)
+        if (_kafkaOptionConfig.EnableAuthorization)
         {
             adminClientConfig.SecurityProtocol = SecurityProtocol.SaslPlaintext;
             adminClientConfig.SaslMechanism = SaslMechanism.Plain;
-            adminClientConfig.SaslUsername = configuration["Kafka:SaslUserName"];
+            adminClientConfig.SaslUsername = _kafkaOptionConfig.SaslUserName;
-            adminClientConfig.SaslPassword = configuration["Kafka:SaslPassword"];
+            adminClientConfig.SaslPassword = _kafkaOptionConfig.SaslUserName;
         }
         return new AdminClientBuilder(adminClientConfig).Build();
     }

modules/JiShe.CollectBus.Kafka/Internal/KafkaPollyPipeline.cs

@@ -34,7 +34,8 @@ namespace JiShe.CollectBus.Kafka.Internal
                 ErrorCode.NotCoordinatorForGroup,
                 ErrorCode.NetworkException,
                 ErrorCode.GroupCoordinatorNotAvailable,
-                ErrorCode.InvalidGroupId
+                ErrorCode.InvalidGroupId,
+                ErrorCode.IllegalGeneration
             };
             return ex switch
             {

protocols/JiShe.CollectBus.Protocol.T37612012/AnalysisData/DataStorage.cs

@@ -287,6 +287,20 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
             };
             _runtimeContext.UseTableSessionPool = false; // 使树模型池
             await _dbProvider.InsertAsync(treeData);
+            // 数据帧
+            var treeFrameData = new TreeModelSingleMeasuringEntity<string>()
+            {
+                SystemName = _applicationOptions.SystemType,
+                DeviceId = $"{data.DeviceId}",
+                DeviceType = $"{data.DeviceType}.{IOTDBDataType.Status}",
+                ProjectId = $"{data.ProjectId}",
+                Timestamps = timestamps,
+                SingleMeasuring = new Tuple<string, string>(ConcentratorStatusFieldConst.FrameData, analysisBaseDto.HexMessage ?? string.Empty)
+            };
+
+            _runtimeContext.UseTableSessionPool = false; // 使树模型池
+            await _dbProvider.InsertAsync(treeFrameData);
+
             // 时间
             var treeRecordingTimeData = new TreeModelSingleMeasuringEntity<long>()
             {
@@ -295,7 +309,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
                 DeviceType = $"{data.DeviceType}.{IOTDBDataType.Status}",
                 ProjectId = $"{data.ProjectId}",
                 Timestamps = timestamps,
-                SingleMeasuring = new Tuple<string, long>(ConcentratorStatusFieldConst.RecordingTime, data.TimeSpan!.Value.GetDateTimeOffset().ToUnixTimeNanoseconds())
+                SingleMeasuring = new Tuple<string, long>(ConcentratorStatusFieldConst.RecordingTime, (data.TimeSpan.HasValue ? data.TimeSpan.Value : DateTime.Now).GetDateTimeOffset().ToUnixTimeNanoseconds())
             };
             _runtimeContext.UseTableSessionPool = false; // 使树模型池
             await _dbProvider.InsertAsync(treeRecordingTimeData);
@@ -311,19 +325,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
             };
             _runtimeContext.UseTableSessionPool = false; // 使树模型池
             await _dbProvider.InsertAsync(treeRemarkData);
-            // 数据帧
-            var treeFrameData = new TreeModelSingleMeasuringEntity<string>()
-            {
-                SystemName = _applicationOptions.SystemType,
-                DeviceId = $"{data.DeviceId}",
-                DeviceType = $"{data.DeviceType}.{IOTDBDataType.Status}",
-                ProjectId = $"{data.ProjectId}",
-                Timestamps = timestamps,
-                SingleMeasuring = new Tuple<string, string>(ConcentratorStatusFieldConst.FrameData, analysisBaseDto.HexMessage ?? string.Empty )
-            };
            
-            _runtimeContext.UseTableSessionPool = false; // 使树模型池
-            await _dbProvider.InsertAsync(treeFrameData);
             return await Task.FromResult(true);
         }
 

services/JiShe.CollectBus.Application.Contracts/Subscribers/IWorkerSubscriberAppService.cs

@@ -38,7 +38,7 @@ namespace JiShe.CollectBus.Subscribers
         /// 电表自动阀控下行消息消费订阅
         /// </summary>
         /// <returns></returns>
-        Task<ISubscribeAck> AmmeterScheduledAutoValveControl(MeterReadingTelemetryPacketInfo issuedEventMessage);
+        Task<ISubscribeAck> AmmeterScheduledAutoValveControl(List<MeterReadingTelemetryPacketInfo> issuedEventMessage);
         #endregion
 
         #region 水表消息采集

services/JiShe.CollectBus.Application/Subscribers/WorkerSubscriberAppService.cs

@@ -5,6 +5,7 @@ using JiShe.CollectBus.Kafka.Attributes;
 using JiShe.CollectBus.Kafka.Internal;
 using Microsoft.Extensions.Logging;
 using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 using TouchSocket.Sockets;
 
@@ -67,11 +68,10 @@ namespace JiShe.CollectBus.Subscribers
         /// </summary>
         /// <param name="receivedMessage"></param>
         /// <returns></returns>
-        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName)]
+        //[KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName)]
         public async Task<ISubscribeAck> AmmeterScheduledMeterFifteenMinuteReadingIssuedEvent(MeterReadingTelemetryPacketInfo receivedMessage)
         {
             _logger.LogInformation("15分钟采集电表数据下行消息消费队列开始处理");
-
             return await SendMessagesAsync(receivedMessage);
         }
 
@@ -80,12 +80,14 @@ namespace JiShe.CollectBus.Subscribers
         /// </summary>
         /// <param name="receivedMessage"></param>
         /// <returns></returns>
-        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName)]
+        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName,EnableBatch =true,TaskCount=30,BatchSize =500)]
-        public async Task<ISubscribeAck> AmmeterScheduledAutoValveControl(MeterReadingTelemetryPacketInfo receivedMessage)
+        public async Task<ISubscribeAck> AmmeterScheduledAutoValveControl(List<MeterReadingTelemetryPacketInfo> receivedMessage)
         {
             //todo 如果是时段自动阀控，需要检查当前的时间，如果时间在自动阀控时间段内，则发送自动阀控报文，否则不发送，尤其是消息队列阻塞或者延时过长的时候。以免造成生产事故。
-            _logger.LogInformation("电表自动阀控下行消息消费队列开始处理");
+            //_logger.LogInformation("电表自动阀控下行消息消费队列开始处理");
-            return await SendMessagesAsync(receivedMessage);
+            _logger.LogWarning($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");
+            return SubscribeAck.Success();
+            //return await SendMessagesAsync(receivedMessage);
         }
 
         /// <summary>

web/JiShe.CollectBus.Host/Pages/Monitor.cshtml

@@ -16,6 +16,7 @@
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <link href="libs/bootstrap/css/bootstrap.min.css" rel="stylesheet" />
     <title>后端服务</title>
+
 </head>
    
 <body>