use concurrentDic

Synapse/EventServer.cs

@@ -18,6 +18,8 @@ public partial class Synapse
     private void RunEventServer()
     {
         Client!.ApplicationMessageReceivedAsync += async e =>
+        {
+            await Task.Run(async () =>
             {
                 if (!e.ApplicationMessage.Topic.StartsWith(EventServerTopicPrefix)) return;
                 var reqBody = e.ApplicationMessage.ConvertPayloadToString();
@@ -37,8 +39,8 @@ public partial class Synapse
                             {
                                 var pt = mt.GetParameters()[1].ParameterType;
                                 mt.Invoke(callClass, pt == typeof(string)
-                                ? new object?[] { eventName, reqBody }
-                                : new object?[] { eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption) });
+                                    ? [eventName, reqBody]
+                                    : [eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption)]);
                             }
                             else
                             {
@@ -52,6 +54,7 @@ public partial class Synapse
                     }))
                     .ToList();
                 await Task.WhenAll(tasks);
+            });
         };
         SubEventServerTopic();
     }

Synapse/RpcClient.cs

@@ -1,20 +1,19 @@
 using System;
 using System.Collections.Generic;
-using System.Linq;
 using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using MQTTnet;
-using MQTTnet.Packets;
 using SimApi.Communications;
 using SimApi.Helpers;
+using System.Collections.Concurrent;
 
 namespace SimApi;
 
 public partial class Synapse
 {
-    private Dictionary<string, TaskCompletionSource<string>> ResponseCompletionSources { get; } = new();
+    private ConcurrentDictionary<string, TaskCompletionSource<string>> ResponseCompletionSources { get; } = new();
 
     private string EventClientTopicPrefix => $"{Options.SysName}/{Options.AppName}/rpc/client/{Options.AppId}/";
 
@@ -27,7 +26,7 @@ public partial class Synapse
             var messageId = e.ApplicationMessage.Topic.Replace(EventClientTopicPrefix, string.Empty);
             if (!ResponseCompletionSources.TryGetValue(messageId, out var tcs)) return Task.CompletedTask;
             tcs.SetResult(reqBody);
-            ResponseCompletionSources.Remove(messageId);
+            ResponseCompletionSources.Remove(messageId, out _);
             return Task.CompletedTask;
         };
         SubRpcClientTopic();
@@ -43,6 +42,7 @@ public partial class Synapse
     private string? FireRpc(string app, string action, object? param, Dictionary<string, string>? headers = null,
         int? timeout = null)
     {
+        // 移除锁语句
         string paramJson;
         if (param is string strParam)
         {
@@ -56,7 +56,7 @@ public partial class Synapse
         var topic = $"{Options.SysName}/{app}/rpc/server/{action}";
         var messageId = Guid.NewGuid().ToString();
         var tcs = new TaskCompletionSource<string>();
-        ResponseCompletionSources.Add(messageId, tcs);
+        ResponseCompletionSources.TryAdd(messageId, tcs);
         var messageBuilder = new MqttApplicationMessageBuilder()
             .WithTopic(topic)
             .WithPayload(paramJson)

Synapse/RpcServer.cs

@@ -20,9 +20,11 @@ public partial class Synapse
 
     private void RunRpcServer()
     {
-        Client!.ApplicationMessageReceivedAsync += e =>
+        Client!.ApplicationMessageReceivedAsync += async e =>
         {
-            if (!e.ApplicationMessage.Topic.StartsWith(RpcServerTopicPrefix)) return Task.CompletedTask;
+            await Task.Run(() =>
+            {
+                if (!e.ApplicationMessage.Topic.StartsWith(RpcServerTopicPrefix)) return;
                 var reqBody = e.ApplicationMessage.ConvertPayloadToString();
                 var action = e.ApplicationMessage.Topic.Replace(RpcServerTopicPrefix, string.Empty);
                 var appInfo = e.ApplicationMessage.ResponseTopic.Split(",");
@@ -58,7 +60,10 @@ public partial class Synapse
                                 var pt2 = mt.GetParameters()[0].ParameterType;
                                 var param2 = pt2 == typeof(string)
                                     ? [reqBody]
-                                : new[] { JsonSerializer.Deserialize(reqBody, pt2, SimApiUtil.JsonOption), headerData };
+                                    : new[]
+                                    {
+                                        JsonSerializer.Deserialize(reqBody, pt2, SimApiUtil.JsonOption), headerData
+                                    };
                                 ret = mt.Invoke(callClass, param2);
                                 break;
                             default:
@@ -98,13 +103,12 @@ public partial class Synapse
                     .WithPayload(returnJson)
                     .WithRetainFlag(false)
                     .Build();
-            if (!Client.IsConnected) return Task.CompletedTask;
+                if (!Client.IsConnected) return;
                 Client.PublishAsync(message, CancellationToken.None).Wait();
                 logger.LogDebug(
                     "Synapse Rpc Server Return: ({BasicPropertiesMessageId}) {BasicPropertiesType}@{OptionsAppName} -> {BasicPropertiesReplyTo}\n{ReturnJson}",
                     appInfo[2], action, Options.AppName, appInfo[0], returnJson);
-
-            return Task.CompletedTask;
+            });
         };
         SubRpcServerTopic();
     }