diff --git a/src/ThingsGateway.Gateway.Application/Plugin/Collect/CollectBase.cs b/src/ThingsGateway.Gateway.Application/Plugin/Collect/CollectBase.cs
index 218521737..f6edfc418 100644
--- a/src/ThingsGateway.Gateway.Application/Plugin/Collect/CollectBase.cs
+++ b/src/ThingsGateway.Gateway.Application/Plugin/Collect/CollectBase.cs
@@ -50,6 +50,8 @@ public abstract class CollectBase : DriverBase
 
     private IStringLocalizer Localizer { get; set; }
 
+    public virtual bool IsSingleThread => true;
+
     internal override void Init(DeviceRunTime device)
     {
         Localizer = App.CreateLocalizerByType(typeof(CollectBase))!;
@@ -89,8 +91,13 @@ public abstract class CollectBase : DriverBase
     {
         try
         {
-            //if (IsSingleThread)
-            //    await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+            if (IsSingleThread)
+            {
+                while (WriteLock.IsWaitting)
+                {
+                    await Task.Delay(100).ConfigureAwait(false);
+                }
+            }
 
             if (cancellationToken.IsCancellationRequested)
                 return new(new OperationCanceledException());
@@ -112,8 +119,6 @@ public abstract class CollectBase : DriverBase
         }
         finally
         {
-            //if (IsSingleThread)
-            //    WriteLock.Release();
         }
     }
 
@@ -126,8 +131,8 @@ public abstract class CollectBase : DriverBase
         try
         {
             // 如果是单线程模式，则等待写入锁
-            //if (IsSingleThread)
-            //    await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+            if (IsSingleThread)
+                await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false);
 
             // 检查协议是否为空，如果为空则抛出异常
             if (Protocol == null)
@@ -139,11 +144,18 @@ public abstract class CollectBase : DriverBase
             // 使用并发方式遍历写入信息列表，并进行异步写入操作
             await writeInfoLists.ParallelForEachAsync(async (writeInfo, cancellationToken) =>
             {
-                // 调用协议的写入方法，将写入信息中的数据写入到对应的寄存器地址，并获取操作结果
-                var result = await Protocol.WriteAsync(writeInfo.Key.RegisterAddress, writeInfo.Value, writeInfo.Key.DataType, cancellationToken).ConfigureAwait(false);
+                try
+                {
+                    // 调用协议的写入方法，将写入信息中的数据写入到对应的寄存器地址，并获取操作结果
+                    var result = await Protocol.WriteAsync(writeInfo.Key.RegisterAddress, writeInfo.Value, writeInfo.Key.DataType, cancellationToken).ConfigureAwait(false);
 
-                // 将操作结果添加到结果字典中，使用变量名称作为键
-                operResults.TryAdd(writeInfo.Key.Name, result);
+                    // 将操作结果添加到结果字典中，使用变量名称作为键
+                    operResults.TryAdd(writeInfo.Key.Name, result);
+                }
+                catch (Exception ex)
+                {
+                    operResults.TryAdd(writeInfo.Key.Name, new(ex));
+                }
             }, CollectProperties.ConcurrentCount, cancellationToken).ConfigureAwait(false);
 
             // 返回包含操作结果的字典
@@ -152,8 +164,8 @@ public abstract class CollectBase : DriverBase
         finally
         {
             // 如果是单线程模式，则释放写入锁
-            //if (IsSingleThread)
-            //    WriteLock.Release();
+            if (IsSingleThread)
+                WriteLock.Release();
         }
     }
 
@@ -535,14 +547,10 @@ public abstract class CollectBase : DriverBase
     /// <param name="isRead">指示是否为读取操作</param>
     /// <param name="cancellationToken">取消操作的通知</param>
     /// <returns>操作结果，包含执行方法的结果</returns>
-    internal async ValueTask<OperResult<object>> InvokeMethodAsync(VariableMethod variableMethod, string? value = null, bool isRead = true, CancellationToken cancellationToken = default)
+    protected virtual async ValueTask<OperResult<object>> InvokeMethodAsync(VariableMethod variableMethod, string? value = null, bool isRead = true, CancellationToken cancellationToken = default)
     {
         try
         {
-            // 如果配置为单线程模式，则获取写入锁定
-            //if (IsSingleThread)
-            //    await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false);
-
             // 初始化操作结果
             OperResult<object> result = new OperResult<object>();
 
@@ -590,9 +598,6 @@ public abstract class CollectBase : DriverBase
         }
         finally
         {
-            // 如果配置为单线程模式，则释放写入锁定
-            //if (IsSingleThread)
-            //    WriteLock.Release();
         }
     }
 
@@ -640,5 +645,77 @@ public abstract class CollectBase : DriverBase
         return results.Concat(results1).ToDictionary(a => a.Key, a => a.Value);
     }
 
+    /// <summary>
+    /// 异步写入方法
+    /// </summary>
+    /// <param name="writeInfoLists">要写入的变量及其对应的数据</param>
+    /// <param name="cancellationToken">取消操作的通知</param>
+    /// <returns>写入操作的结果字典</returns>
+    internal async ValueTask<Dictionary<string, OperResult<object>>> InvokeMethodAsync(Dictionary<VariableRunTime, JToken> writeInfoLists, CancellationToken cancellationToken)
+    {
+        // 初始化结果字典
+        Dictionary<string, OperResult<object>> results = new Dictionary<string, OperResult<object>>();
+
+        // 遍历写入信息列表
+        foreach (var (deviceVariable, jToken) in writeInfoLists)
+        {
+            // 检查是否有写入表达式
+            if (!string.IsNullOrEmpty(deviceVariable.WriteExpressions))
+            {
+                // 提取原始数据
+                object rawdata = jToken is JValue jValue ? jValue.Value : jToken is JArray jArray ? jArray : jToken.ToString();
+                try
+                {
+                    // 根据写入表达式转换数据
+                    object data = deviceVariable.WriteExpressions.GetExpressionsResult(rawdata);
+                    // 将转换后的数据重新赋值给写入信息列表
+                    writeInfoLists[deviceVariable] = JToken.FromObject(data);
+                }
+                catch (Exception ex)
+                {
+                    // 如果转换失败，则记录错误信息
+                    results.Add(deviceVariable.Name, new OperResult<object>(Localizer["WriteExpressionsError", deviceVariable.Name, deviceVariable.WriteExpressions, ex.Message], ex));
+                }
+            }
+        }
+
+        ConcurrentDictionary<string, OperResult<object>> operResults = new();
+
+        try
+        {
+            // 如果是单线程模式，则等待写入锁
+            if (IsSingleThread)
+                await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+
+            // 使用并发方式遍历写入信息列表，并进行异步写入操作
+            await writeInfoLists
+            .Where(a => !results.Any(b => b.Key == a.Key.Name))
+            .ToDictionary(item => item.Key, item => item.Value).ParallelForEachAsync(async (writeInfo, cancellationToken) =>
+        {
+            try
+            {
+                // 调用协议的写入方法，将写入信息中的数据写入到对应的寄存器地址，并获取操作结果
+                var result = await InvokeMethodAsync(writeInfo.Key.VariableMethod, writeInfo.Value?.ToString(), false, cancellationToken).ConfigureAwait(false);
+
+                // 将操作结果添加到结果字典中，使用变量名称作为键
+                operResults.TryAdd(writeInfo.Key.Name, result);
+            }
+            catch (Exception ex)
+            {
+                operResults.TryAdd(writeInfo.Key.Name, new(ex));
+            }
+        }, CollectProperties.ConcurrentCount, cancellationToken).ConfigureAwait(false);
+        }
+        finally
+        {
+            // 如果是单线程模式，则释放写入锁
+            if (IsSingleThread)
+                WriteLock.Release();
+        }
+
+        // 将转换失败的变量和写入成功的变量的操作结果合并到结果字典中
+        return results.Concat(operResults).ToDictionary(a => a.Key, a => a.Value);
+    }
+
     #endregion 写入方法
 }
diff --git a/src/ThingsGateway.Gateway.Application/Services/Rpc/RpcService.cs b/src/ThingsGateway.Gateway.Application/Services/Rpc/RpcService.cs
index 07cabd0a7..cad16fe0a 100644
--- a/src/ThingsGateway.Gateway.Application/Services/Rpc/RpcService.cs
+++ b/src/ThingsGateway.Gateway.Application/Services/Rpc/RpcService.cs
@@ -45,7 +45,7 @@ public class RpcService : IRpcService
     {
         // 初始化用于存储将要写入的变量和方法的字典
         Dictionary<CollectBase, Dictionary<VariableRunTime, JToken>> WriteVariables = new();
-        Dictionary<CollectBase, Dictionary<VariableRunTime, string>> WriteMethods = new();
+        Dictionary<CollectBase, Dictionary<VariableRunTime, JToken>> WriteMethods = new();
         // 用于存储结果的并发字典
         ConcurrentDictionary<string, OperResult> results = new();
 
@@ -107,15 +107,16 @@ public class RpcService : IRpcService
                 }
                 else
                 {
+                    JToken tagValue = JTokenUtil.GetJTokenFromString(item.Value);
                     // 执行方法
                     if (WriteMethods.ContainsKey(dev))
                     {
-                        WriteMethods[dev].Add(tag, item.Value);
+                        WriteMethods[dev].Add(tag, tagValue);
                     }
                     else
                     {
                         WriteMethods.Add(dev, new());
-                        WriteMethods[dev].Add(tag, item.Value);
+                        WriteMethods[dev].Add(tag, tagValue);
                     }
                 }
             }
@@ -183,41 +184,49 @@ public class RpcService : IRpcService
         // 使用并行方式执行方法
         await WriteMethods.ParallelForEachAsync(async (item, cancellationToken) =>
         {
-            await item.Value.ParallelForEachAsync(async (writeMethod, cancellationToken) =>
+            try
             {
-                // 执行变量附带的方法
-                var method = item.Key.CurrentDevice.VariableMethods.FirstOrDefault(it => it.Variable == writeMethod.Key);
-                OperResult<object> result;
-                try
-                {
-                    result = await item.Key.InvokeMethodAsync(method, writeMethod.Value, false, cancellationToken);
-                }
-                catch (Exception ex)
-                {
-                    result = new OperResult<object>(ex);
-                }
+                // 调用设备的写入方法
+                var result = await item.Key.InvokeMethodAsync(item.Value, cancellationToken);
+
+                Dictionary<string, string> operateMethods = item.Value.Select(a => a.Key).ToDictionary(a => a.Name, a => a.OtherMethod!);
 
                 // 写入日志
-                _logQueues.Enqueue(
-                    new RpcLog()
+                foreach (var resultItem in result)
+                {
+                    // 写入日志
+                    _logQueues.Enqueue(
+                        new RpcLog()
+                        {
+                            LogTime = DateTime.Now,
+                            OperateMessage = resultItem.Value.IsSuccess ? null : resultItem.Value.ToString(),
+                            IsSuccess = resultItem.Value.IsSuccess,
+                            OperateMethod = operateMethods[resultItem.Key],
+                            OperateObject = resultItem.Key,
+                            OperateSource = sourceDes,
+                            ParamJson = items[resultItem.Key]?.ToString(),
+                            ResultJson = resultItem.Value.Content?.ToString()
+                        }
+                    );
+
+                    // 不返回详细错误
+                    if (!resultItem.Value.IsSuccess)
                     {
-                        LogTime = DateTime.Now,
-                        OperateMessage = result.IsSuccess ? null : result.ToString(),
-                        IsSuccess = result.IsSuccess,
-                        OperateMethod = writeMethod.Key.OtherMethod!,
-                        OperateObject = writeMethod.Key.Name,
-                        OperateSource = sourceDes,
-                        ParamJson = writeMethod.Value?.ToString(),
-                        ResultJson = result.Content?.ToString()
+                        OperResult<object> result1 = resultItem.Value;
+                        result1.Exception = null;
+                        result[resultItem.Key] = result1;
                     }
-                );
-
-                // 不返回详细错误
-                result.Exception = null;
-                results.TryAdd(writeMethod.Key.Name, result);
-            }, item.Key.CollectProperties.ConcurrentCount, cancellationToken);
+                }
+            }
+            catch (Exception ex)
+            {
+                // 将异常信息添加到结果字典中
+                results.AddRange(item.Value.Select((KeyValuePair<VariableRunTime, JToken> a) =>
+                {
+                    return new KeyValuePair<string, OperResult>(a.Key.Name, new OperResult(ex));
+                }));
+            }
         }, Environment.ProcessorCount / 2, cancellationToken);
-
         // 返回结果字典
         return new(results);
     }
diff --git a/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.cs b/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.cs
index c355f7d0b..1e5627893 100644
--- a/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.cs
+++ b/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.cs
@@ -14,6 +14,7 @@ using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 
 using MQTTnet.AspNetCore;
+using MQTTnet.Server;
 
 using ThingsGateway.Foundation;
 
@@ -81,6 +82,8 @@ public partial class MqttServer : BusinessBaseWithCacheIntervalScript<VariableDa
         });
         if (_mqttServer != null)
         {
+            _mqttServer.ClientDisconnectedAsync -= MqttServer_ClientDisconnectedAsync;
+            _mqttServer.ClientDisconnectedAsync += MqttServer_ClientDisconnectedAsync;
             _mqttServer.ValidatingConnectionAsync -= MqttServer_ValidatingConnectionAsync;
             _mqttServer.InterceptingPublishAsync -= MqttServer_InterceptingPublishAsync;
             _mqttServer.LoadingRetainedMessageAsync -= MqttServer_LoadingRetainedMessageAsync;
@@ -104,6 +107,7 @@ public partial class MqttServer : BusinessBaseWithCacheIntervalScript<VariableDa
     {
         if (_mqttServer != null)
         {
+            _mqttServer.ClientDisconnectedAsync -= MqttServer_ClientDisconnectedAsync;
             _mqttServer.ValidatingConnectionAsync -= MqttServer_ValidatingConnectionAsync;
             _mqttServer.InterceptingPublishAsync -= MqttServer_InterceptingPublishAsync;
             _mqttServer.LoadingRetainedMessageAsync -= MqttServer_LoadingRetainedMessageAsync;
diff --git a/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.other.cs b/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.other.cs
index b1d6f26b8..f0fb1efa9 100644
--- a/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.other.cs
+++ b/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.other.cs
@@ -183,6 +183,12 @@ public partial class MqttServer : BusinessBaseWithCacheIntervalScript<VariableDa
         await CompletedTask.Instance;
     }
 
+    private Task MqttServer_ClientDisconnectedAsync(MQTTnet.Server.ClientDisconnectedEventArgs arg)
+    {
+        LogMessage?.LogInformation($"{ToString()}-{arg.ClientId}-Client DisConnected-{arg.DisconnectType}");
+        return Task.CompletedTask;
+    }
+
     private async Task MqttServer_ValidatingConnectionAsync(ValidatingConnectionEventArgs arg)
     {
         if (!arg.ClientId.StartsWith(_driverPropertys.StartWithId))