修改代码

优化登录心跳逻辑

modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs

@@ -7,17 +7,20 @@ using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using System.Collections.Concurrent;
 using System.Text;
-using Volo.Abp.Timing;
+using System.Text.RegularExpressions;
 
 namespace JiShe.CollectBus.Kafka.Consumer
 {
     public class ConsumerService : IConsumerService, IDisposable
     {
         private readonly ILogger<ConsumerService> _logger;
-        private readonly ConcurrentDictionary<Type, (object Consumer, CancellationTokenSource CTS)>
+        /// <summary>
+        /// 消费者存储
+        /// Key 格式：{groupId}_{topic}_{TKey}_{TValue}
+        /// </summary>
+        private readonly ConcurrentDictionary<string, (object Consumer, CancellationTokenSource CTS)>
             _consumerStore = new();
         private readonly KafkaOptionConfig _kafkaOptionConfig;
-        private class KafkaConsumer<TKey, TValue> where TKey : notnull where TValue : class { }
 
         /// <summary>
         /// ConsumerService
@@ -109,15 +112,14 @@ namespace JiShe.CollectBus.Kafka.Consumer
         /// <returns></returns>
         public async Task SubscribeAsync<TKey, TValue>(string[] topics, Func<TKey, TValue, Task<bool>> messageHandler, string? groupId = null) where TKey : notnull where TValue : class
         {
-            var consumerKey = typeof(KafkaConsumer<TKey, TValue>);
+            var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(TKey).Name}_{typeof(TValue).Name}";
             var cts = new CancellationTokenSource();
 
-            //var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
-            //(
-            //    CreateConsumer<TKey, TValue>(groupId),
-            //    cts
-            //)).Consumer as IConsumer<TKey, TValue>;
-            var consumer = CreateConsumer<TKey, TValue>(groupId);
+            var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
+            (
+                CreateConsumer<TKey, TValue>(groupId),
+                cts
+            )).Consumer as IConsumer<TKey, TValue>;
             consumer!.Subscribe(topics);
 
             await Task.Run(async () =>
@@ -177,19 +179,14 @@ namespace JiShe.CollectBus.Kafka.Consumer
         public async Task SubscribeAsync<TValue>(string[] topics, Func<TValue, Task<bool>> messageHandler, string? groupId) where TValue : class
         {
             try {
-                var consumerKey = typeof(KafkaConsumer<Ignore, TValue>);
+                var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(Ignore).Name}_{typeof(TValue).Name}";
                 var cts = new CancellationTokenSource();
-                //if (topics.Contains(ProtocolConst.SubscriberLoginReceivedEventName))
-                //{
-                //    string ssss = "";
-                //}
-                //var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
-                //(
-                //    CreateConsumer<string, TValue>(groupId),
-                //    cts
-                //)).Consumer as IConsumer<string, TValue>;
+                var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
+                (
+                    CreateConsumer<Ignore, TValue>(groupId),
+                    cts
+                )).Consumer as IConsumer<Ignore, TValue>;
 
-                var consumer = CreateConsumer<Ignore, TValue>(groupId);
                 consumer!.Subscribe(topics);
 
                 _ = Task.Run(async () =>
@@ -275,15 +272,14 @@ namespace JiShe.CollectBus.Kafka.Consumer
         /// <param name="batchTimeout">批次超时时间</param>
         public async Task SubscribeBatchAsync<TKey, TValue>(string[] topics,Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null,int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class
         {
-            var consumerKey = typeof(KafkaConsumer<TKey, TValue>);
+            var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(TKey).Name}_{typeof(TValue).Name}";
             var cts = new CancellationTokenSource();
 
-            //var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
-            //    (
-            //        CreateConsumer<TKey, TValue>(groupId),
-            //        cts
-            //    )).Consumer as IConsumer<TKey, TValue>;
-            var consumer = CreateConsumer<string, TValue>(groupId);
+            var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
+                (
+                    CreateConsumer<TKey, TValue>(groupId),
+                    cts
+                )).Consumer as IConsumer<TKey, TValue>;
             consumer!.Subscribe(topics);
 
             var timeout = batchTimeout ?? TimeSpan.FromSeconds(5); // 默认超时时间调整为5秒
@@ -367,6 +363,7 @@ namespace JiShe.CollectBus.Kafka.Consumer
                     catch (OperationCanceledException)
                     {
                         // 任务取消，正常退出
+                        
                     }
                     catch (Exception ex)
                     {
@@ -408,16 +405,15 @@ namespace JiShe.CollectBus.Kafka.Consumer
         /// <param name="consumeTimeout">消费等待时间</param>
         public async Task SubscribeBatchAsync<TValue>(string[] topics,Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100,TimeSpan? batchTimeout = null,TimeSpan? consumeTimeout = null)where TValue : class
         {
-            var consumerKey = typeof(KafkaConsumer<string, TValue>);
+            var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(Ignore).Name}_{typeof(TValue).Name}";
             var cts = new CancellationTokenSource();
 
-            //var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
-            //    (
-            //        CreateConsumer<string, TValue>(groupId),
-            //        cts
-            //    )).Consumer as IConsumer<string, TValue>;
+            var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
+                (
+                    CreateConsumer<Ignore, TValue>(groupId),
+                    cts
+                )).Consumer as IConsumer<Ignore, TValue>;
 
-            var consumer= CreateConsumer<string, TValue> (groupId);
             consumer!.Subscribe(topics);
 
             var timeout = batchTimeout ?? TimeSpan.FromSeconds(5); // 默认超时时间调整为5秒
@@ -519,9 +515,9 @@ namespace JiShe.CollectBus.Kafka.Consumer
         /// </summary>
         /// <typeparam name="TKey"></typeparam>
         /// <typeparam name="TValue"></typeparam>
-        public void Unsubscribe<TKey, TValue>() where TKey : notnull where TValue : class
+        public void Unsubscribe<TKey, TValue>(string[] topics, string groupId) where TKey : notnull where TValue : class
         {
-            var consumerKey = typeof((TKey, TValue));
+            var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(Ignore).Name}_{typeof(TValue).Name}";
             if (_consumerStore.TryRemove(consumerKey, out var entry))
             {
                 entry.CTS.Cancel();

modules/JiShe.CollectBus.Kafka/Consumer/IConsumerService.cs

@@ -46,5 +46,5 @@ public interface IConsumerService
         string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null)
         where TValue : class;
 
-    void Unsubscribe<TKey, TValue>() where TKey : notnull where TValue : class;
+    void Unsubscribe<TKey, TValue>(string[] topics, string groupId) where TKey : notnull where TValue : class;
 }

protocols/JiShe.CollectBus.Protocol.Contracts/AnalysisStrategyContext.cs

@@ -9,11 +9,9 @@ using System.Threading.Tasks;
 namespace JiShe.CollectBus.Protocol.Contracts
 {
 
-    public class AnalysisStrategyContext
+    public class AnalysisStrategyContext(IServiceProvider provider)
     {
-        private readonly IServiceProvider _provider;
-
-        public AnalysisStrategyContext(IServiceProvider provider) => _provider = provider;
+        private readonly IServiceProvider _provider = provider;
 
         /// <summary>
         /// 执行策略

protocols/JiShe.CollectBus.Protocol.Contracts/Protocol/Dto/AFN0_F1_AnalysisDto.cs

@@ -8,6 +8,6 @@ namespace JiShe.CollectBus.Protocol.Dto
 {
     public class AFN0_F1_AnalysisDto: UnitDataDto
     {
-
+        public bool Verify { get; set; } = true;
     }
 }

protocols/JiShe.CollectBus.Protocol/AnalysisData/AFN_00H/AFN0_F1_Analysis.cs

@@ -12,6 +12,9 @@ using System.Threading.Tasks;
 
 namespace JiShe.CollectBus.Protocol.AnalysisData.AFN_00H
 {
+    /// <summary>
+    /// 全部确认：对收到报文中的全部数据单元标识进行确认  
+    /// </summary>
     public class AFN0_F1_Analysis: IAnalysisStrategy<TB3761, AFN0_F1_AnalysisDto>
     {
         private readonly ILogger<AFN0_F1_Analysis> _logger;

protocols/JiShe.CollectBus.Protocol/StandardProtocolPlugin.cs

@@ -28,16 +28,18 @@ namespace JiShe.CollectBus.Protocol
         private readonly IProducerService _producerService;
 
         private readonly IRepository<Device, Guid> _deviceRepository;
+        private readonly ITcpService _tcpService;
         /// <summary>
         /// Initializes a new instance of the <see cref="StandardProtocolPlugin"/> class.
         /// </summary>
         /// <param name="serviceProvider">The service provider.</param>
-        public StandardProtocolPlugin(IServiceProvider serviceProvider,ILogger<StandardProtocolPlugin> logger) : base(serviceProvider, logger)
+        public StandardProtocolPlugin(IServiceProvider serviceProvider,ILogger<StandardProtocolPlugin> logger, ITcpService tcpService) : base(serviceProvider, logger)
         {
-            _logger= logger;
+            _logger = logger;
             //_logger = serviceProvider.GetRequiredService<ILogger<StandardProtocolPlugin>>();
             _producerService = serviceProvider.GetRequiredService<IProducerService>();
             _deviceRepository = serviceProvider.GetRequiredService<IRepository<Device, Guid>>();
+            _tcpService = tcpService;
         }
 
         public sealed override ProtocolInfo Info => new(nameof(StandardProtocolPlugin), "376.1", "TCP", "376.1协议", "DTS1980");
@@ -146,9 +148,21 @@ namespace JiShe.CollectBus.Protocol
                 Fn = 1
             };
             var bytes = Build3761SendData.BuildSendCommandBytes(reqParam);
+            var issuedEventMessage = new IssuedEventMessage 
+            { 
+                ClientId = messageReceivedLoginEvent.ClientId, 
+                DeviceNo = messageReceivedLoginEvent.DeviceNo, 
+                Message = bytes, Type = IssuedEventType.Login,
+                MessageId = messageReceivedLoginEvent.MessageId
+            };
             //await _producerBus.PublishAsync(ProtocolConst.SubscriberLoginIssuedEventName, new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceived.MessageId });
+            if (_tcpService.ClientExists(issuedEventMessage.ClientId))
+            {
+                await _tcpService.SendAsync(issuedEventMessage.ClientId, issuedEventMessage.Message);
+                _logger.LogInformation($"集中器地址{issuedEventMessage.ClientId} 登录回复下发内容:{Convert.ToHexString(bytes)}");
+                await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginIssuedEventName, issuedEventMessage);
+            }
 
-            await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginIssuedEventName, new IssuedEventMessage { ClientId = messageReceivedLoginEvent.ClientId, DeviceNo = messageReceivedLoginEvent.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceivedLoginEvent.MessageId });
             //await _producerBus.Publish(new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceived.MessageId });
         }
 
@@ -226,11 +240,24 @@ namespace JiShe.CollectBus.Protocol
             };
             var bytes = Build3761SendData.BuildSendCommandBytes(reqParam);
             //await _producerBus.PublishAsync(ProtocolConst.SubscriberHeartbeatIssuedEventName, new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceived.MessageId });
-
-            await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatIssuedEventName, new IssuedEventMessage { ClientId = messageReceivedHeartbeatEvent.ClientId, DeviceNo = messageReceivedHeartbeatEvent.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceivedHeartbeatEvent.MessageId });
+ 
+            IssuedEventMessage issuedEventMessage = new IssuedEventMessage
+            {
+                ClientId = messageReceivedHeartbeatEvent.ClientId,
+                DeviceNo = messageReceivedHeartbeatEvent.DeviceNo,
+                Message = bytes,
+                Type = IssuedEventType.Heartbeat,
+                MessageId = messageReceivedHeartbeatEvent.MessageId
+            };
+            if (_tcpService.ClientExists(issuedEventMessage.ClientId))
+            {
+                await _tcpService.SendAsync(issuedEventMessage.ClientId, bytes);
+                _logger.LogWarning($"集中器地址{issuedEventMessage.ClientId} 心跳回复下发内容:{Convert.ToHexString(bytes)}");
+                await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatIssuedEventName, issuedEventMessage);
+            }
 
             //await _producerBus.Publish(new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceived.MessageId });
-            
+
         }
 
 

services/JiShe.CollectBus.Application/Samples/SampleAppService.cs

@@ -168,7 +168,7 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
         {
             SystemName = "energy",
             DeviceId = "402440506",
-            DeviceType = "Ammeter",
+            DeviceType = "1",
             ProjectId = "10059",
             Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(),
             SingleMeasuring = new Tuple<string, string>(measuring, value)

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

@@ -76,14 +76,11 @@ namespace JiShe.CollectBus.Subscribers
                     isAck=false;
                     break;
                 }
-                _logger.LogInformation($"集中器地址{issuedEventMessage.ClientId} 登录回复下发内容:{issuedEventMessage.Serialize()}");
-
+                
                 loginEntity.AckTime = Clock.Now;
                 loginEntity.IsAck = true;
                 await _messageReceivedLoginEventRepository.UpdateAsync(loginEntity);
-                if (_tcpService.ClientExists(issuedEventMessage.ClientId))
-                    await _tcpService.SendAsync(issuedEventMessage.ClientId, issuedEventMessage.Message);
-                isAck = true;
+               isAck = true;
 
             }
             // TODO:暂时ACK，等后续处理是否放到私信队列中
@@ -102,19 +99,10 @@ namespace JiShe.CollectBus.Subscribers
                     isAck = false;
                     break;
                 }
-                _logger.LogWarning($"集中器地址{issuedEventMessage.ClientId} 心跳回复下发内容:{issuedEventMessage.Serialize()}");
-
                 heartbeatEntity.AckTime = Clock.Now;
                 heartbeatEntity.IsAck = true;
                 await _messageReceivedHeartbeatEventRepository.UpdateAsync(heartbeatEntity);
-                //var device = await _deviceRepository.FindAsync(a => a.Number == issuedEventMessage.DeviceNo);
-                //if (device != null)
-                //{
-                //    await _tcpService.SendAsync(device.ClientId, issuedEventMessage.Message);
-                //}
-                if(_tcpService.ClientExists(issuedEventMessage.ClientId))
-                    await _tcpService.SendAsync(issuedEventMessage.ClientId, issuedEventMessage.Message);
-            }
+             }
             // TODO:暂时ACK，等后续处理是否放到私信队列中
             return isAck ? SubscribeAck.Success() : SubscribeAck.Fail();
         }
@@ -176,19 +164,6 @@ namespace JiShe.CollectBus.Subscribers
         [KafkaSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName, EnableBatch = true)]
         public async Task<ISubscribeAck> ReceivedHeartbeatEvent(List<MessageReceivedHeartbeat> receivedHeartbeatMessages)
         {
-            //foreach (var receivedHeartbeatMessage in receivedHeartbeatMessages)
-            //{
-            //    var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
-            //    if (protocolPlugin == null)
-            //    {
-            //        _logger.LogError("协议不存在！");
-            //    }
-            //    else
-            //    {
-            //        //await protocolPlugin.HeartbeatAsync(receivedHeartbeatMessage);
-            //        await _messageReceivedHeartbeatEventRepository.InsertAsync(receivedHeartbeatMessage);
-            //    }
-            //}
             await _messageReceivedHeartbeatEventRepository.InsertManyAsync(receivedHeartbeatMessages);
 
             return SubscribeAck.Success();
@@ -197,20 +172,6 @@ namespace JiShe.CollectBus.Subscribers
         [KafkaSubscribe(ProtocolConst.SubscriberLoginReceivedEventName,EnableBatch =true)]
         public async Task<ISubscribeAck> ReceivedLoginEvent(List<MessageReceivedLogin> receivedLoginMessages)
         {
-            //foreach (var receivedLoginMessage in receivedLoginMessages)
-            //{
-                //var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
-                //if (protocolPlugin == null)
-                //{
-                //    _logger.LogError("协议不存在！");
-                //}
-                //else
-                //{
-                //    //await protocolPlugin.LoginAsync(receivedLoginMessage);
-                //    await _messageReceivedLoginEventRepository.InsertAsync(receivedLoginMessage);
-                //}
-             
-            //}
             await _messageReceivedLoginEventRepository.InsertManyAsync(receivedLoginMessages);
             return SubscribeAck.Success();
         }

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

@@ -16,7 +16,6 @@
     <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>