feat(ITcpSocketClient): add concurrency connect logic (#6377)

ArgoZhang · web-flow · commit e3d081edd936 · 2025-07-08T15:51:43.000+08:00
* feat: 增加锁控制并发

* refactor: 合并重连方法提高可读性

* feat: 使用 SemaphoreSlim 信号量控制重连并发逻辑

* test: 增加重连单元测试
diff --git a/src/BootstrapBlazor/Services/TcpSocket/TcpSocketClientBase.cs b/src/BootstrapBlazor/Services/TcpSocket/TcpSocketClientBase.cs
@@ -65,6 +65,8 @@ public abstract class TcpSocketClientBase(SocketClientOptions options) : ITcpSoc
     private CancellationTokenSource? _receiveCancellationTokenSource;
     private CancellationTokenSource? _autoConnectTokenSource;
 
+    private readonly SemaphoreSlim _semaphoreSlim = new(1, 1);
+
     /// <summary>
     /// <inheritdoc/>
     /// </summary>
@@ -78,49 +80,72 @@ public async ValueTask<bool> ConnectAsync(IPEndPoint endPoint, CancellationToken
             return true;
         }
 
+        var connectionToken = GenerateConnectionToken(token);
+        try
+        {
+            await _semaphoreSlim.WaitAsync(connectionToken).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException)
+        {
+            // 如果信号量等待被取消，则直接返回 IsConnected
+            // 不管是超时还是被取消，都不需要重连，肯定有其他线程在连接中
+            return IsConnected;
+        }
+
+        if (IsConnected)
+        {
+            _semaphoreSlim.Release();
+            return true;
+        }
+
+        var reconnect = true;
         var ret = false;
         SocketClientProvider = ServiceProvider?.GetRequiredService<ISocketClientProvider>()
             ?? throw new InvalidOperationException("SocketClientProvider is not registered in the service provider.");
 
         try
         {
-            ret = await ConnectCoreAsync(SocketClientProvider, endPoint, token);
-            if (ret == false)
-            {
-                Reconnect();
-            }
+            ret = await ConnectCoreAsync(SocketClientProvider, endPoint, connectionToken);
         }
         catch (OperationCanceledException ex)
         {
             if (token.IsCancellationRequested)
             {
                 Log(LogLevel.Warning, ex, $"TCP Socket connect operation was canceled from {LocalEndPoint} to {endPoint}");
+                reconnect = false;
             }
             else
             {
                 Log(LogLevel.Warning, ex, $"TCP Socket connect operation timed out from {LocalEndPoint} to {endPoint}");
-                Reconnect();
             }
         }
         catch (Exception ex)
         {
             Log(LogLevel.Error, ex, $"TCP Socket connection failed from {LocalEndPoint} to {endPoint}");
+        }
+
+        // 释放信号量
+        _semaphoreSlim.Release();
+
+        if (!ret && reconnect)
+        {
             Reconnect();
         }
+
         return ret;
     }
 
     private void Reconnect()
     {
-        if (options.IsAutoReconnect)
+        if (options.IsAutoReconnect && _remoteEndPoint != null)
         {
             Task.Run(async () =>
             {
                 try
                 {
                     _autoConnectTokenSource ??= new();
                     await Task.Delay(options.ReconnectInterval, _autoConnectTokenSource.Token).ConfigureAwait(false);
-                    HandleDisconnection();
+                    await ConnectAsync(_remoteEndPoint, _autoConnectTokenSource.Token).ConfigureAwait(false);
                 }
                 catch { }
             }, CancellationToken.None).ConfigureAwait(false);
@@ -138,14 +163,7 @@ private async ValueTask<bool> ConnectCoreAsync(ISocketClientProvider provider, I
         _localEndPoint = Options.LocalEndPoint;
         _remoteEndPoint = endPoint;
 
-        var connectionToken = token;
-        if (Options.ConnectTimeout > 0)
-        {
-            // 设置连接超时时间
-            var connectTokenSource = new CancellationTokenSource(options.ConnectTimeout);
-            connectionToken = CancellationTokenSource.CreateLinkedTokenSource(token, connectTokenSource.Token).Token;
-        }
-        var ret = await provider.ConnectAsync(endPoint, connectionToken);
+        var ret = await provider.ConnectAsync(endPoint, token);
 
         if (ret)
         {
@@ -159,13 +177,16 @@ private async ValueTask<bool> ConnectCoreAsync(ISocketClientProvider provider, I
         return ret;
     }
 
-    private void HandleDisconnection()
+    private CancellationToken GenerateConnectionToken(CancellationToken token)
     {
-        if (options.IsAutoReconnect && _remoteEndPoint != null)
+        var connectionToken = token;
+        if (Options.ConnectTimeout > 0)
         {
-            _autoConnectTokenSource ??= new();
-            _ = Task.Run(() => ConnectAsync(_remoteEndPoint, _autoConnectTokenSource.Token)).ConfigureAwait(false);
+            // 设置连接超时时间
+            var connectTokenSource = new CancellationTokenSource(options.ConnectTimeout);
+            connectionToken = CancellationTokenSource.CreateLinkedTokenSource(token, connectTokenSource.Token).Token;
         }
+        return connectionToken;
     }
 
     /// <summary>
@@ -182,6 +203,7 @@ public virtual async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, Cancel
         }
 
         var ret = false;
+        var reconnect = true;
         try
         {
             var sendToken = token;
@@ -192,15 +214,12 @@ public virtual async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, Cancel
                 sendToken = CancellationTokenSource.CreateLinkedTokenSource(token, sendTokenSource.Token).Token;
             }
             ret = await SocketClientProvider.SendAsync(data, sendToken);
-            if (ret == false)
-            {
-                HandleDisconnection();
-            }
         }
         catch (OperationCanceledException ex)
         {
             if (token.IsCancellationRequested)
             {
+                reconnect = false;
                 Log(LogLevel.Warning, ex, $"TCP Socket send operation was canceled from {_localEndPoint} to {_remoteEndPoint}");
             }
             else
@@ -211,14 +230,18 @@ public virtual async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, Cancel
         catch (Exception ex)
         {
             Log(LogLevel.Error, ex, $"TCP Socket send failed from {_localEndPoint} to {_remoteEndPoint}");
-
-            HandleDisconnection();
         }
 
         if (options.EnableLog)
         {
             Log(LogLevel.Information, null, $"Sending data from {_localEndPoint} to {_remoteEndPoint}, Data Length: {data.Length} Data Content: {BitConverter.ToString(data.ToArray())} Result: {ret}");
         }
+
+        if (!ret && reconnect)
+        {
+            // 如果发送失败并且需要重连则尝试重连
+            Reconnect();
+        }
         return ret;
     }
 
@@ -244,7 +267,7 @@ public virtual async ValueTask<Memory<byte>> ReceiveAsync(CancellationToken toke
         var len = await ReceiveCoreAsync(SocketClientProvider, buffer, token);
         if (len == 0)
         {
-            HandleDisconnection();
+            Reconnect();
         }
         return buffer[..len];
     }
@@ -270,11 +293,12 @@ private async ValueTask AutoReceiveAsync()
             }
         }
 
-        HandleDisconnection();
+        Reconnect();
     }
 
     private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memory<byte> buffer, CancellationToken token)
     {
+        var reconnect = true;
         var len = 0;
         try
         {
@@ -292,8 +316,6 @@ private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memo
                 // 远端主机关闭链路
                 Log(LogLevel.Information, null, $"TCP Socket {_localEndPoint} received 0 data closed by {_remoteEndPoint}");
                 buffer = Memory<byte>.Empty;
-
-                HandleDisconnection();
             }
             else
             {
@@ -311,6 +333,7 @@ private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memo
             if (token.IsCancellationRequested)
             {
                 Log(LogLevel.Warning, ex, $"TCP Socket receive operation canceled from {_localEndPoint} to {_remoteEndPoint}");
+                reconnect = false;
             }
             else
             {
@@ -320,14 +343,18 @@ private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memo
         catch (Exception ex)
         {
             Log(LogLevel.Error, ex, $"TCP Socket receive failed from {_localEndPoint} to {_remoteEndPoint}");
-
-            HandleDisconnection();
         }
 
         if (options.EnableLog)
         {
             Log(LogLevel.Information, null, $"Receiving data from {_localEndPoint} to {_remoteEndPoint}, Data Length: {len} Data Content: {BitConverter.ToString(buffer.ToArray())}");
         }
+
+        if (len == 0 && reconnect)
+        {
+            // 如果接收数据长度为 0 并且需要重连则尝试重连
+            Reconnect();
+        }
         return len;
     }
 
diff --git a/test/UnitTest/Services/TcpSocketFactoryTest.cs b/test/UnitTest/Services/TcpSocketFactoryTest.cs
@@ -69,6 +69,18 @@ public async Task ConnectAsync_Cancel()
         var cst = new CancellationTokenSource();
         cst.Cancel();
         var connect = await client.ConnectAsync("localhost", 9999, cst.Token);
+
+        // 由于信号量被取消，所以连接会失败
+        Assert.False(connect);
+
+        // 测试真正的连接被取消逻辑
+        cst = new CancellationTokenSource();
+        _ = Task.Run(async () =>
+        {
+            await Task.Delay(50);
+            cst.Cancel();
+        });
+        connect = await client.ConnectAsync("localhost", 9999, cst.Token);
         Assert.False(connect);
     }
 
@@ -102,6 +114,29 @@ public async Task ConnectAsync_Error()
         methodInfo.Invoke(client, [LogLevel.Error, null!, "Test error log"]);
     }
 
+    [Fact]
+    public async Task ConnectAsync_Lock()
+    {
+        // 测试并发锁问题
+        var provider = new MockAutoReconnectLockSocketProvider();
+        var client = CreateClient(builder =>
+        {
+            builder.AddTransient<ISocketClientProvider>(p => provider);
+        });
+
+        // 开 5 个线程同时连接
+        _ = Task.Run(async () =>
+        {
+            // 延时 150 保证有一个连接失败
+            await Task.Delay(150);
+            provider.SetConnected(true);
+        });
+        var results = await Task.WhenAll(Enumerable.Range(1, 5).Select(i => client.ConnectAsync("localhost", 0).AsTask()));
+        // 期望结果是 1个 false 4个 true
+        Assert.Equal(1, results.Count(r => !r));
+        Assert.Equal(4, results.Count(r => r));
+    }
+
     [Fact]
     public async Task Send_Timeout()
     {
@@ -432,6 +467,7 @@ public async Task AutoReconnect_Cancel()
         });
 
         await client.ConnectAsync("localhost", 0);
+        await Task.Delay(100);
         await client.DisposeAsync();
     }
 
@@ -912,6 +948,41 @@ public async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, CancellationTo
         }
     }
 
+    class MockAutoReconnectLockSocketProvider : ISocketClientProvider
+    {
+        public bool IsConnected { get; private set; }
+
+        public IPEndPoint LocalEndPoint { get; set; } = new IPEndPoint(IPAddress.Loopback, 0);
+
+        public async ValueTask<bool> ConnectAsync(IPEndPoint endPoint, CancellationToken token = default)
+        {
+            await Task.Delay(100, token);
+            return IsConnected;
+        }
+
+        public ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, CancellationToken token = default)
+        {
+            return ValueTask.FromResult(true);
+        }
+
+        public ValueTask<int> ReceiveAsync(Memory<byte> buffer, CancellationToken token = default)
+        {
+            byte[] data = [1, 2, 3, 4, 5];
+            data.CopyTo(buffer);
+            return ValueTask.FromResult(5);
+        }
+
+        public ValueTask CloseAsync()
+        {
+            return ValueTask.CompletedTask;
+        }
+
+        public void SetConnected(bool state)
+        {
+            IsConnected = state;
+        }
+    }
+
     class MockAutoReconnectSocketProvider : ISocketClientProvider
     {
         public bool IsConnected { get; private set; }

Original file line number	Diff line number	Diff line change
`@@ -65,6 +65,8 @@ public abstract class TcpSocketClientBase(SocketClientOptions options) : ITcpSoc`
`65`	`65`	`private CancellationTokenSource? _receiveCancellationTokenSource;`
`66`	`66`	`private CancellationTokenSource? _autoConnectTokenSource;`
`67`	`67`
	`68`	`+ private readonly SemaphoreSlim _semaphoreSlim = new(1, 1);`
	`69`	`+`
`68`	`70`	`/// <summary>`
`69`	`71`	`/// <inheritdoc/>`
`70`	`72`	`/// </summary>`
`@@ -78,49 +80,72 @@ public async ValueTask<bool> ConnectAsync(IPEndPoint endPoint, CancellationToken`
`78`	`80`	`return true;`
`79`	`81`	`}`
`80`	`82`
	`83`	`+ var connectionToken = GenerateConnectionToken(token);`
	`84`	`+ try`
	`85`	`+ {`
	`86`	`+ await _semaphoreSlim.WaitAsync(connectionToken).ConfigureAwait(false);`
	`87`	`+ }`
	`88`	`+ catch (OperationCanceledException)`
	`89`	`+ {`
	`90`	`+ // 如果信号量等待被取消，则直接返回 IsConnected`
	`91`	`+ // 不管是超时还是被取消，都不需要重连，肯定有其他线程在连接中`
	`92`	`+ return IsConnected;`
	`93`	`+ }`
	`94`	`+`
	`95`	`+ if (IsConnected)`
	`96`	`+ {`
	`97`	`+ _semaphoreSlim.Release();`
	`98`	`+ return true;`
	`99`	`+ }`
	`100`	`+`
	`101`	`+ var reconnect = true;`
`81`	`102`	`var ret = false;`
`82`	`103`	`SocketClientProvider = ServiceProvider?.GetRequiredService<ISocketClientProvider>()`
`83`	`104`	`?? throw new InvalidOperationException("SocketClientProvider is not registered in the service provider.");`
`84`	`105`
`85`	`106`	`try`
`86`	`107`	`{`
`87`		`- ret = await ConnectCoreAsync(SocketClientProvider, endPoint, token);`
`88`		`- if (ret == false)`
`89`		`- {`
`90`		`- Reconnect();`
`91`		`- }`
	`108`	`+ ret = await ConnectCoreAsync(SocketClientProvider, endPoint, connectionToken);`
`92`	`109`	`}`
`93`	`110`	`catch (OperationCanceledException ex)`
`94`	`111`	`{`
`95`	`112`	`if (token.IsCancellationRequested)`
`96`	`113`	`{`
`97`	`114`	`Log(LogLevel.Warning, ex, $"TCP Socket connect operation was canceled from {LocalEndPoint} to {endPoint}");`
	`115`	`+ reconnect = false;`
`98`	`116`	`}`
`99`	`117`	`else`
`100`	`118`	`{`
`101`	`119`	`Log(LogLevel.Warning, ex, $"TCP Socket connect operation timed out from {LocalEndPoint} to {endPoint}");`
`102`		`- Reconnect();`
`103`	`120`	`}`
`104`	`121`	`}`
`105`	`122`	`catch (Exception ex)`
`106`	`123`	`{`
`107`	`124`	`Log(LogLevel.Error, ex, $"TCP Socket connection failed from {LocalEndPoint} to {endPoint}");`
	`125`	`+ }`
	`126`	`+`
	`127`	`+ // 释放信号量`
	`128`	`+ _semaphoreSlim.Release();`
	`129`	`+`
	`130`	`+ if (!ret && reconnect)`
	`131`	`+ {`
`108`	`132`	`Reconnect();`
`109`	`133`	`}`
	`134`	`+`
`110`	`135`	`return ret;`
`111`	`136`	`}`
`112`	`137`
`113`	`138`	`private void Reconnect()`
`114`	`139`	`{`
`115`		`- if (options.IsAutoReconnect)`
	`140`	`+ if (options.IsAutoReconnect && _remoteEndPoint != null)`
`116`	`141`	`{`
`117`	`142`	`Task.Run(async () =>`
`118`	`143`	`{`
`119`	`144`	`try`
`120`	`145`	`{`
`121`	`146`	`_autoConnectTokenSource ??= new();`
`122`	`147`	`await Task.Delay(options.ReconnectInterval, _autoConnectTokenSource.Token).ConfigureAwait(false);`
`123`		`- HandleDisconnection();`
	`148`	`+ await ConnectAsync(_remoteEndPoint, _autoConnectTokenSource.Token).ConfigureAwait(false);`
`124`	`149`	`}`
`125`	`150`	`catch { }`
`126`	`151`	`}, CancellationToken.None).ConfigureAwait(false);`
`@@ -138,14 +163,7 @@ private async ValueTask<bool> ConnectCoreAsync(ISocketClientProvider provider, I`
`138`	`163`	`_localEndPoint = Options.LocalEndPoint;`
`139`	`164`	`_remoteEndPoint = endPoint;`
`140`	`165`
`141`		`- var connectionToken = token;`
`142`		`- if (Options.ConnectTimeout > 0)`
`143`		`- {`
`144`		`- // 设置连接超时时间`
`145`		`- var connectTokenSource = new CancellationTokenSource(options.ConnectTimeout);`
`146`		`- connectionToken = CancellationTokenSource.CreateLinkedTokenSource(token, connectTokenSource.Token).Token;`
`147`		`- }`
`148`		`- var ret = await provider.ConnectAsync(endPoint, connectionToken);`
	`166`	`+ var ret = await provider.ConnectAsync(endPoint, token);`
`149`	`167`
`150`	`168`	`if (ret)`
`151`	`169`	`{`
`@@ -159,13 +177,16 @@ private async ValueTask<bool> ConnectCoreAsync(ISocketClientProvider provider, I`
`159`	`177`	`return ret;`
`160`	`178`	`}`
`161`	`179`
`162`		`- private void HandleDisconnection()`
	`180`	`+ private CancellationToken GenerateConnectionToken(CancellationToken token)`
`163`	`181`	`{`
`164`		`- if (options.IsAutoReconnect && _remoteEndPoint != null)`
	`182`	`+ var connectionToken = token;`
	`183`	`+ if (Options.ConnectTimeout > 0)`
`165`	`184`	`{`
`166`		`- _autoConnectTokenSource ??= new();`
`167`		`- _ = Task.Run(() => ConnectAsync(_remoteEndPoint, _autoConnectTokenSource.Token)).ConfigureAwait(false);`
	`185`	`+ // 设置连接超时时间`
	`186`	`+ var connectTokenSource = new CancellationTokenSource(options.ConnectTimeout);`
	`187`	`+ connectionToken = CancellationTokenSource.CreateLinkedTokenSource(token, connectTokenSource.Token).Token;`
`168`	`188`	`}`
	`189`	`+ return connectionToken;`
`169`	`190`	`}`
`170`	`191`
`171`	`192`	`/// <summary>`
`@@ -182,6 +203,7 @@ public virtual async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, Cancel`
`182`	`203`	`}`
`183`	`204`
`184`	`205`	`var ret = false;`
	`206`	`+ var reconnect = true;`
`185`	`207`	`try`
`186`	`208`	`{`
`187`	`209`	`var sendToken = token;`
`@@ -192,15 +214,12 @@ public virtual async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, Cancel`
`192`	`214`	`sendToken = CancellationTokenSource.CreateLinkedTokenSource(token, sendTokenSource.Token).Token;`
`193`	`215`	`}`
`194`	`216`	`ret = await SocketClientProvider.SendAsync(data, sendToken);`
`195`		`- if (ret == false)`
`196`		`- {`
`197`		`- HandleDisconnection();`
`198`		`- }`
`199`	`217`	`}`
`200`	`218`	`catch (OperationCanceledException ex)`
`201`	`219`	`{`
`202`	`220`	`if (token.IsCancellationRequested)`
`203`	`221`	`{`
	`222`	`+ reconnect = false;`
`204`	`223`	`Log(LogLevel.Warning, ex, $"TCP Socket send operation was canceled from {_localEndPoint} to {_remoteEndPoint}");`
`205`	`224`	`}`
`206`	`225`	`else`
`@@ -211,14 +230,18 @@ public virtual async ValueTask<bool> SendAsync(ReadOnlyMemory<byte> data, Cancel`
`211`	`230`	`catch (Exception ex)`
`212`	`231`	`{`
`213`	`232`	`Log(LogLevel.Error, ex, $"TCP Socket send failed from {_localEndPoint} to {_remoteEndPoint}");`
`214`		`-`
`215`		`- HandleDisconnection();`
`216`	`233`	`}`
`217`	`234`
`218`	`235`	`if (options.EnableLog)`
`219`	`236`	`{`
`220`	`237`	`Log(LogLevel.Information, null, $"Sending data from {_localEndPoint} to {_remoteEndPoint}, Data Length: {data.Length} Data Content: {BitConverter.ToString(data.ToArray())} Result: {ret}");`
`221`	`238`	`}`
	`239`	`+`
	`240`	`+ if (!ret && reconnect)`
	`241`	`+ {`
	`242`	`+ // 如果发送失败并且需要重连则尝试重连`
	`243`	`+ Reconnect();`
	`244`	`+ }`
`222`	`245`	`return ret;`
`223`	`246`	`}`
`224`	`247`
`@@ -244,7 +267,7 @@ public virtual async ValueTask<Memory<byte>> ReceiveAsync(CancellationToken toke`
`244`	`267`	`var len = await ReceiveCoreAsync(SocketClientProvider, buffer, token);`
`245`	`268`	`if (len == 0)`
`246`	`269`	`{`
`247`		`- HandleDisconnection();`
	`270`	`+ Reconnect();`
`248`	`271`	`}`
`249`	`272`	`return buffer[..len];`
`250`	`273`	`}`
`@@ -270,11 +293,12 @@ private async ValueTask AutoReceiveAsync()`
`270`	`293`	`}`
`271`	`294`	`}`
`272`	`295`
`273`		`- HandleDisconnection();`
	`296`	`+ Reconnect();`
`274`	`297`	`}`
`275`	`298`
`276`	`299`	`private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memory<byte> buffer, CancellationToken token)`
`277`	`300`	`{`
	`301`	`+ var reconnect = true;`
`278`	`302`	`var len = 0;`
`279`	`303`	`try`
`280`	`304`	`{`
`@@ -292,8 +316,6 @@ private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memo`
`292`	`316`	`// 远端主机关闭链路`
`293`	`317`	`Log(LogLevel.Information, null, $"TCP Socket {_localEndPoint} received 0 data closed by {_remoteEndPoint}");`
`294`	`318`	`buffer = Memory<byte>.Empty;`
`295`		`-`
`296`		`- HandleDisconnection();`
`297`	`319`	`}`
`298`	`320`	`else`
`299`	`321`	`{`
`@@ -311,6 +333,7 @@ private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memo`
`311`	`333`	`if (token.IsCancellationRequested)`
`312`	`334`	`{`
`313`	`335`	`Log(LogLevel.Warning, ex, $"TCP Socket receive operation canceled from {_localEndPoint} to {_remoteEndPoint}");`
	`336`	`+ reconnect = false;`
`314`	`337`	`}`
`315`	`338`	`else`
`316`	`339`	`{`
`@@ -320,14 +343,18 @@ private async ValueTask<int> ReceiveCoreAsync(ISocketClientProvider client, Memo`
`320`	`343`	`catch (Exception ex)`
`321`	`344`	`{`
`322`	`345`	`Log(LogLevel.Error, ex, $"TCP Socket receive failed from {_localEndPoint} to {_remoteEndPoint}");`
`323`		`-`
`324`		`- HandleDisconnection();`
`325`	`346`	`}`
`326`	`347`
`327`	`348`	`if (options.EnableLog)`
`328`	`349`	`{`
`329`	`350`	`Log(LogLevel.Information, null, $"Receiving data from {_localEndPoint} to {_remoteEndPoint}, Data Length: {len} Data Content: {BitConverter.ToString(buffer.ToArray())}");`
`330`	`351`	`}`
	`352`	`+`
	`353`	`+ if (len == 0 && reconnect)`
	`354`	`+ {`
	`355`	`+ // 如果接收数据长度为 0 并且需要重连则尝试重连`
	`356`	`+ Reconnect();`
	`357`	`+ }`
`331`	`358`	`return len;`
`332`	`359`	`}`
`333`	`360`