细节调整

Ys.Scada.Backend/Controllers/HomeController.cs

@@ -1,15 +0,0 @@
-using Microsoft.AspNetCore.Authorization;
-using Microsoft.AspNetCore.Mvc;
-
-namespace Ys.Scada.Backend.Controllers
-{
-    //public class HomeController : Controller
-    //{
-    //    [Route("{*url}", Order = 999)]
-    //    [AllowAnonymous]
-    //    public IActionResult Index()
-    //    {
-    //        return File("index.html", "text/html");
-    //    }
-    //}
-}

Ys.Scada.Backend/EventConsumers/TagDataConsumerData.cs

@@ -14,21 +14,31 @@ namespace Ys.Scada.Backend.EventConsumers
     public class TagDataConsumerData : IConsumer<TransportMessage>
     {
         private readonly DataServer _dataServer;
-        public TagDataConsumerData(DataServer dataServer) { _dataServer = dataServer; }
+        private readonly DedicatedThreadTaskScheduler _scheduler;
+        public TagDataConsumerData(DataServer dataServer)
+        {
+            _dataServer = dataServer;
+            _scheduler = new DedicatedThreadTaskScheduler(10);
+        }
         public Task HandleEvent(TransportMessage eventMessage)
         {
             var message = Encoding.UTF8.GetString(eventMessage.Body.ToArray());
             if (!string.IsNullOrWhiteSpace(message))
-            { 
+            {
                 var tags = JsonConvert.DeserializeObject<List<TagDataEventModel>>(message);
-                if (tags .Any())
+                if (tags.Any())
                 {
-                    foreach (var tag in tags)
+                    Parallel.ForEach(tags, new ParallelOptions
+                    {
+                        MaxDegreeOfParallelism = 10,
+                        TaskScheduler = _scheduler
+                    },
+                     (tag) =>
                     {
                         var key = Global.GetVariableKey(tag.Id);
                         ForgeCache.Instance.SetValue(key, tag.VL);
-                        _dataServer.MyMulticast(tag);
-                    }
+                        _dataServer.MyMulticast(tag).GetAwaiter().GetResult();
+                    });
                 }
             }
             return Task.CompletedTask;

Ys.Scada.Backend/Startup.cs

@@ -1,33 +1,22 @@
-using Microsoft.AspNetCore.Mvc.ModelBinding.Binders;
+using Microsoft.AspNetCore.Authentication.JwtBearer;
 using Microsoft.AspNetCore.Mvc.ModelBinding;
-using Microsoft.AspNetCore.Mvc;
-using Microsoft.Extensions.Configuration;
+using Microsoft.AspNetCore.Mvc.ModelBinding.Binders;
+using Microsoft.Extensions.FileProviders;
+using Microsoft.IdentityModel.Tokens;
 using Microsoft.OpenApi.Models;
-using Newtonsoft.Json.Serialization;
-using Newtonsoft.Json.Converters;
 using Newtonsoft.Json;
-using Newtonsoft.Json.Linq;
-using Microsoft.AspNetCore.Authentication.JwtBearer;
-using Microsoft.IdentityModel.Tokens;
 using System.Text;
-using Ys.Scada.Core;
-using Ys.Scada.Backend.Authorization;
 using System.Text.Json.Serialization;
-using Microsoft.Extensions.FileProviders;
+using System.Threading.Channels;
+using Ys.Scada.Backend.Authorization;
+using Ys.Scada.Backend.EventConsumers;
+using Ys.Scada.Backend.WS;
+using Ys.Scada.Core;
 using Ys.Scada.Core.EventBus.Consumer;
 using Ys.Scada.Core.EventBus.Publisher;
 using Ys.Scada.Core.EventBus.Subscriber;
-using Ys.Scada.Core.EventModels;
-using Ys.Scada.Backend.EventConsumers;
-using Ys.Scada.Manager;
-using Ys.Scada.Backend.WS;
-using Microsoft.Extensions.DependencyInjection.Extensions;
-using static Ys.Scada.Backend.Controllers.RealTimeDataController;
-using System.Threading.Channels;
-using Microsoft.Extensions.DependencyInjection;
-using Ys.Scada.Core.Common;
-using Microsoft.AspNetCore.Hosting;
 using Ys.Scada.Core.MQ;
+using Ys.Scada.Manager;
 
 namespace Ys.Scada.Backend
 {
@@ -156,9 +145,7 @@ namespace Ys.Scada.Backend
             app.UseAuthorization();
             app.UseStaticFiles(new StaticFileOptions
             {
-                FileProvider = new PhysicalFileProvider(
-                Path.Combine(env.ContentRootPath, "pic")),
-                RequestPath = "/pic"
+                FileProvider = new PhysicalFileProvider( Path.Combine(env.ContentRootPath, "pic")),  RequestPath = "/pic"
             }); 
             app.UseSwagger();
             app.UseSwaggerUI(c =>

Ys.Scada.Backend/WS/DataServer.cs

@@ -27,15 +27,21 @@ namespace Ys.Scada.Backend.WS
             _clients = new ConcurrentDictionary<string, DataSession>();
             _secretKey = secretKey;
         }
-        public async Task MyMulticast(TagDataEventModel tag)
+        public  Task MyMulticast(TagDataEventModel tag)
         {
             foreach (var client in _clients.Values)
             {
-                var ukey = Global.GetUserKey(client.userId.ToString());
-                var codes = ForgeCache.Instance.GetValue<string>(ukey);
-                if (codes.Contains(tag.Id))
-                    client.SendData(tag);
+                if (client.IsConnected)
+                {
+                    var ukey = Global.GetUserKey(client.userId.ToString());
+                    var codes = ForgeCache.Instance.GetValue<string>(ukey);
+                    if (codes.Contains(tag.Id))
+                        client.SendData(tag);
+                }
+                else
+                    RemoveClient(client.Id.ToString());
             }
+            return Task.CompletedTask;
         }
         protected override TcpSession CreateSession() 
         {

Ys.Scada.Backend/WS/TagDataChannelConsumer.cs

@@ -38,13 +38,13 @@ namespace Ys.Scada.Backend.WS
         public TagDataChannelConsumer(MemoryChannel<TransportMessage> channel, IEventPublisher eventPublisher, string count)
         {
             _channel = channel;
-            handlerCount = int.TryParse(count, out int c)  ? c : 25;
+            handlerCount = int.TryParse(count, out int c)  ? c : 50;
             _eventPublisher = eventPublisher;
-            scheduler = new DedicatedThreadTaskScheduler(50);
+            scheduler = new DedicatedThreadTaskScheduler(handlerCount);
         }
         public Task Excute()
         {
-            Enumerable.Range(1, 50).ToList().ForEach(_ =>
+            Enumerable.Range(1, handlerCount).ToList().ForEach(_ =>
             {
                 Task.Factory.StartNew(async () =>
                 {

Ys.Scada.Kafka/KafkaConsumerClient.cs

@@ -93,6 +93,8 @@ namespace Ys.Scada.Kafka
         }
         public void Dispose()
         {
+            _consumerClient?.Close();
+            _consumerClient?.Dispose();
         }
     }
 }

Ys.Scada.MQTT/MQTTConsumerClient.cs

@@ -6,6 +6,7 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
+using Ys.Scada.Core;
 using Ys.Scada.Core.MQ;
 
 namespace Ys.Scada.MQTT
@@ -17,9 +18,15 @@ namespace Ys.Scada.MQTT
         public Func<TransportMessage, object?, Task>? OnMessageCallback { get; set; }
         private MqttClient _consumerClient;
         public BrokerAddress BrokerAddress => new("RabbitMQ", $"{_mqttOptions.Server}:{_mqttOptions.Port}");
-        public MQTTConsumerClient(IOptions<MQTTOptions> options)
+        private readonly MemoryChannel<TransportMessage> _memoryChannel;
+        private int _id;
+        private readonly Producer<TransportMessage> _producer;
+        public MQTTConsumerClient(IOptions<MQTTOptions> options, MemoryChannel<TransportMessage> memoryChannel, int id)
         {
             _mqttOptions = options.Value;
+            _memoryChannel = memoryChannel;
+            _producer = new Producer<TransportMessage>(_memoryChannel.Channel.Writer);
+            _id = id;
         }
         public void Connect()
         {
@@ -32,7 +39,7 @@ namespace Ys.Scada.MQTT
                     var mqttFactory = new MqttFactory();
                     _consumerClient = mqttFactory.CreateMqttClient() as MqttClient;
                     var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer(_mqttOptions.Server, _mqttOptions.Port).Build();
-                     _consumerClient!.ConnectAsync(mqttClientOptions, CancellationToken.None).GetAwaiter().GetResult();
+                    _consumerClient!.ConnectAsync(mqttClientOptions, CancellationToken.None).GetAwaiter().GetResult();
                 }
             }
             finally
@@ -44,13 +51,20 @@ namespace Ys.Scada.MQTT
         {
             Connect();
             _consumerClient.ApplicationMessageReceivedAsync += _consumerClient_ApplicationMessageReceivedAsync;
+            while (true)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                cancellationToken.WaitHandle.WaitOne(timeout);
+            }
         }
         private async Task _consumerClient_ApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
         {
-            var message = new TransportMessage();
-            message.Headers.Add("topic",arg.ApplicationMessage.Topic);
-            message.Headers.Add("clientid", arg.ClientId);
-            await OnMessageCallback!(message,arg);
+            var headers = new Dictionary<string, string?>();
+            headers.Add("topic", arg.ApplicationMessage.Topic);
+            headers.Add("clientid", arg.ClientId);
+            var message = new TransportMessage(headers, arg.ApplicationMessage.Payload);
+
+            await _producer.PublishAsync(message);
         }
         public async void Subscribe(IEnumerable<string> topics)
         {
@@ -60,6 +74,7 @@ namespace Ys.Scada.MQTT
         }
         public void Dispose()
         {
+            _consumerClient?.Dispose();
         }
     }
 }

Ys.Scada.RabbitMQ/RabbitMQConsumerClient.cs

@@ -50,6 +50,7 @@ namespace Ys.Scada.RabbitMQ
             consumer.Received += OnConsumerReceived;
             consumer.Shutdown += OnConsumerShutdown;
             consumer.ConsumerCancelled += OnConsumerConsumerCancelled;
+            
             if (_rabbitMQOptions.BasicQosOptions != null)
                 _channel?.BasicQos(0, _rabbitMQOptions.BasicQosOptions.PrefetchCount, _rabbitMQOptions.BasicQosOptions.Global);
             try
@@ -136,7 +137,6 @@ namespace Ys.Scada.RabbitMQ
         private async Task OnConsumerConsumerCancelled(object? sender, ConsumerEventArgs e)
         {
             await OnStopCallback(_id);
-            //OnLogCallback!(args);
         }
         private async Task OnConsumerReceived(object? sender, BasicDeliverEventArgs e)
         {