use concurrentDic
This commit is contained in:
@@ -1,20 +1,19 @@
|
|||||||
using System;
|
using System;
|
||||||
using System.Collections.Generic;
|
using System.Collections.Generic;
|
||||||
using System.Linq;
|
|
||||||
using System.Text.Json;
|
using System.Text.Json;
|
||||||
using System.Threading;
|
using System.Threading;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using Microsoft.Extensions.Logging;
|
using Microsoft.Extensions.Logging;
|
||||||
using MQTTnet;
|
using MQTTnet;
|
||||||
using MQTTnet.Packets;
|
|
||||||
using SimApi.Communications;
|
using SimApi.Communications;
|
||||||
using SimApi.Helpers;
|
using SimApi.Helpers;
|
||||||
|
using System.Collections.Concurrent;
|
||||||
|
|
||||||
namespace SimApi;
|
namespace SimApi;
|
||||||
|
|
||||||
public partial class Synapse
|
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}/";
|
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);
|
var messageId = e.ApplicationMessage.Topic.Replace(EventClientTopicPrefix, string.Empty);
|
||||||
if (!ResponseCompletionSources.TryGetValue(messageId, out var tcs)) return Task.CompletedTask;
|
if (!ResponseCompletionSources.TryGetValue(messageId, out var tcs)) return Task.CompletedTask;
|
||||||
tcs.SetResult(reqBody);
|
tcs.SetResult(reqBody);
|
||||||
ResponseCompletionSources.Remove(messageId);
|
ResponseCompletionSources.Remove(messageId, out _);
|
||||||
return Task.CompletedTask;
|
return Task.CompletedTask;
|
||||||
};
|
};
|
||||||
SubRpcClientTopic();
|
SubRpcClientTopic();
|
||||||
@@ -43,9 +42,7 @@ public partial class Synapse
|
|||||||
private string? FireRpc(string app, string action, object? param, Dictionary<string, string>? headers = null,
|
private string? FireRpc(string app, string action, object? param, Dictionary<string, string>? headers = null,
|
||||||
int? timeout = null)
|
int? timeout = null)
|
||||||
{
|
{
|
||||||
// 检查并发情况下的线程安全问题,使用锁保证字典操作的原子性
|
// 移除锁语句
|
||||||
lock (ResponseCompletionSources)
|
|
||||||
{
|
|
||||||
string paramJson;
|
string paramJson;
|
||||||
if (param is string strParam)
|
if (param is string strParam)
|
||||||
{
|
{
|
||||||
@@ -59,7 +56,7 @@ public partial class Synapse
|
|||||||
var topic = $"{Options.SysName}/{app}/rpc/server/{action}";
|
var topic = $"{Options.SysName}/{app}/rpc/server/{action}";
|
||||||
var messageId = Guid.NewGuid().ToString();
|
var messageId = Guid.NewGuid().ToString();
|
||||||
var tcs = new TaskCompletionSource<string>();
|
var tcs = new TaskCompletionSource<string>();
|
||||||
ResponseCompletionSources.Add(messageId, tcs);
|
ResponseCompletionSources.TryAdd(messageId, tcs);
|
||||||
var messageBuilder = new MqttApplicationMessageBuilder()
|
var messageBuilder = new MqttApplicationMessageBuilder()
|
||||||
.WithTopic(topic)
|
.WithTopic(topic)
|
||||||
.WithPayload(paramJson)
|
.WithPayload(paramJson)
|
||||||
@@ -102,5 +99,4 @@ public partial class Synapse
|
|||||||
|
|
||||||
return response;
|
return response;
|
||||||
}
|
}
|
||||||
}
|
|
||||||
}
|
}
|
||||||
Reference in New Issue
Block a user