Socket.IO连接随机触发未处理TaskCanceledException求助

云上老周

2026-6-24

问题描述

运行AlgoOrderPlacement.exe时，随机触发未处理的System.Threading.Tasks.TaskCanceledException异常，代码中无主动取消任务的操作，已添加try-catch块和事件处理仍无法定位问题。异常在连接正常运行2-3小时后随机出现，无固定复现步骤。

错误信息

Application: AlgoOrderPlacement.exe
Framework Version: v4.0.30319
Description: The process was terminated due to an unhandled exception.
Exception Info: System.Threading.Tasks.TaskCanceledException
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(System.Threading.Tasks.Task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(System.Threading.Tasks.Task)
at SocketIOClient.Transport.HttpTransport+d__13.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(System.Threading.Tasks.Task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(System.Threading.Tasks.Task)
at SocketIOClient.Transport.TransportRouter+d__43.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(System.Threading.Tasks.Task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(System.Threading.Tasks.Task)
at SocketIOClient.Transport.TransportRouter+d__41.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(System.Threading.Tasks.Task)
at SocketIOClient.SocketIO+d__81.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at SocketIOClient.SocketIO+d__81.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(System.Threading.Tasks.Task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(System.Threading.Tasks.Task)
at SocketIOClient.SocketIO+d__105.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Threading.ExecutionContext.RunInternal(System.Threading.ExecutionContext, System.Threading.ContextCallback, System.Object, Boolean)
at System.Threading.ExecutionContext.Run(System.Threading.ExecutionContext, System.Threading.ContextCallback, System.Object, Boolean)
at System.Threading.QueueUserWorkItemCallback.System.Threading.IThreadPoolWorkItem.ExecuteWorkItem()
at System.Threading.ThreadPoolWorkQueue.Dispatch()

相关代码

namespace AlgoOrderManagement.Helper
{
    public class OrderSocketConnection
    {
        protected SocketIO Socket { get; set; } = null;
        long lastSocketSequence = 0;
        public Dictionary<string, OrderData> OrderStatusDetails = new Dictionary<string, OrderData>();
        private User currentUser = null;

        public delegate void SocketStatus(bool status);
        public event SocketStatus SocketStatusUpdate;
        string URL = string.Empty;
        public void StartSocket(User user, string URL)
        {
            this.currentUser = user;
            this.URL = URL;
            _ = StartSocketAsync();
        }

        private async Task StartSocketAsync()
        {
            try
            {
                if (this.Socket != null)
                {
                    UnSubscribeToConnectionEvents();
                }

                if (this.Socket != null)
                {
                    this.Socket?.DisconnectAsync();
                    this.Socket?.Dispose();
                }

                this.Socket = new SocketIO(URL, new SocketIOOptions
                {
                    Query = new Dictionary<string, string>() { { "token", currentUser.TOKEN }, { "userID", currentUser.USERID }, { "apiType", "INTERACTIVE" } },
                    EIO = 3,
                    Transport = SocketIOClient.Transport.TransportProtocol.Polling,
                    Path = "/interactive/socket.io"
                });

                SubscribeToConnectionEvents();
                FileLogger.logMessage(currentUser.USERNAME + " Before StartSocket");
                await this.Socket.ConnectAsync();
                FileLogger.logMessage(currentUser.USERNAME + " After StartSocket");

            }
            catch (Exception ex)
            {
                FileLogger.logMessage("User StartSocket EX = " + ex.Message + ex.StackTrace);
                await Task.Delay(5000);
                _ = StartSocketAsync();
            }
        }

        private void Socket_OnReconnecting(object sender, int e)
        {
            FileLogger.logMessage(currentUser.USERNAME + " SOCKET SUCCESS= " + e, currentUser.USERNAME);
        }

        private void Socket_OnDisconnected(object sender, string e)
        {
            FileLogger.logMessage(currentUser.USERNAME + " SOCKET DIS-CONNECT= " + e, currentUser.USERNAME);
            UpdateUISocketStatus(false);
        }

        private void Socket_OnPong(object sender, TimeSpan e)
        {
            
        }

        private void Socket_OnPing(object sender, EventArgs e)
        {
            
        }

        private void Socket_OnConnected(object sender, EventArgs e)
        {
            FileLogger.logMessage(currentUser.USERNAME + " SOCKET SUCCESS= " + e, currentUser.USERNAME);
        }

        private void OrderUpdate(object order)
        {
            try
            {
                dynamic data = JObject.Parse(order.ToString());
                if (data != null)
                {
                    if (data.SequenceNumber.Value > lastSocketSequence)
                    {
                        lastSocketSequence = data.SequenceNumber.Value;
                        OrderData orderData = new OrderData();
                        orderData.orderID = data.AppOrderID;
                        orderData.order_status = data.OrderStatus;
                        orderData.cancelRejectReason = data.CancelRejectReason;
                        orderData.order_execute_price = data.OrderAverageTradedPrice;
                        orderData.quantity = data.CumulativeQuantity;
                        orderData.instrument_token = data.ExchangeInstrumentID;
                        orderData.lastUpdateTime = DateTime.Now;
                        if (OrderStatusDetails.ContainsKey(orderData.orderID))
                        {
                            OrderStatusDetails[orderData.orderID] = orderData;
                        }
                        else
                        {
                            OrderStatusDetails.Add(orderData.orderID, orderData);
                        }
                    }

                    CheckCancelOrder(data);
                }
            }
            catch (Exception ex)
            {
                FileLogger.logMessage("OrderUpdate EX = " + ex.Message + ex.StackTrace, currentUser.USERNAME);
            }
        }

        private void CheckCancelOrder(dynamic data)
        {
            try
            {
                string orderStatus = data.OrderStatus.ToString();
                string cancelRejectReason = data.CancelRejectReason.ToString();
                string orderGeneratedDateTimeAPI = data.LastUpdateDateTimeAPI.ToString();
                if (orderStatus.ToLower().Equals("cancelled".ToLower()))
                {
                    TelegramSend.SendError(currentUser.USERNAME + " = Order Cancelled Please Check Once = "
                        + data.CancelRejectReason + Environment.NewLine +
                        "TIME = " + orderGeneratedDateTimeAPI);
                }
            }
            catch (Exception ex)
            {

            }
        }

        private bool SubscribeToConnectionEvents()
        {
            if (this.Socket == null)
                return false;

            this.Socket.On("joined", (data) =>
            {
                FileLogger.logMessage(" SOCKET JOINED= " + data, currentUser.USERNAME);
                UpdateUISocketStatus(true);
            });

            this.Socket.On("success", (data) =>
            {
                FileLogger.logMessage(" SOCKET SUCCESS= " + data, currentUser.USERNAME);
            });

            this.Socket.On("warning", (data) =>
            {
                FileLogger.logMessage(" SOCKET WARNING= " + data, currentUser.USERNAME);
            });

            this.Socket.On("error", (data) =>
            {
                FileLogger.logMessage(" SOCKET ERROR= " + data, currentUser.USERNAME);
                UpdateUISocketStatus(false);
            });

            this.Socket.On("logout", (data) =>
            {
                FileLogger.logMessage(currentUser.USERNAME + " SOCKET LOGOUT= " + data, currentUser.USERNAME);
                UpdateUISocketStatus(false);
            });

            this.Socket.On("order", (order) =>
            {
                OrderUpdate(order.GetValue<string>());
                FileLogger.logMessage(currentUser.USERNAME + " SOCKET ORDER = " + order.ToString(), currentUser.USERNAME);
            });

            this.Socket.On("trade", (trade) =>
            {
                OrderUpdate(trade.GetValue<string>());
                FileLogger.logMessage(currentUser.USERNAME + " SOCKET TRADE = " + trade.ToString(), currentUser.USERNAME);
            });

            this.Socket.OnConnected += Socket_OnConnected;
            this.Socket.OnPing += Socket_OnPing;
            this.Socket.OnPong += Socket_OnPong;
            this.Socket.OnDisconnected += Socket_OnDisconnected;
            this.Socket.OnReconnectAttempt += Socket_OnReconnecting;

            return true;
        }

        private void UnSubscribeToConnectionEvents()
        {
            if (this.Socket != null)
            {
                this.Socket?.Off("logout");
                this.Socket?.Off("error");
                this.Socket?.Off("warning");
                this.Socket?.Off("success");
                this.Socket?.Off("joined");
                this.Socket?.Off("order");
                this.Socket?.Off("trade");
                this.Socket.OnConnected -= Socket_OnConnected;
                this.Socket.OnPing -= Socket_OnPing;
                this.Socket.OnPong -= Socket_OnPong;
                this.Socket.OnDisconnected -= Socket_OnDisconnected;
                this.Socket.OnReconnectAttempt -= Socket_OnReconnecting;
            }
        }

        private void UpdateUISocketStatus(bool status)
        {
            try
            {
                SocketStatusUpdate?.Invoke(status);
                if (status == false)
                {
                    TelegramSend.SendError(currentUser.USERNAME + " Socekt Disconnected ");
                }
            }
            catch (Exception ex)
            {
                FileLogger.logMessage("User UpdateUISocketStatus EX = " + ex.Message + ex.StackTrace, currentUser.USERNAME);
            }
        }
    }
}

问题分析与解决方案

核心原因定位

从调用栈看，异常来自SocketIOClient的HttpTransport层，属于网络层面的任务取消，并非代码主动取消。常见触发场景：

服务器端主动断开长连接（Polling模式下超时无响应）
网络波动导致请求超时
SocketIO客户端内部的超时机制触发任务取消

具体修复措施

1. 捕获并处理`TaskCanceledException`

当前StartSocketAsync的try-catch捕获了Exception，但异常可能在SocketIO客户端内部的异步任务中抛出（未被当前catch覆盖）。需要：

在全局异常处理中添加对TaskCanceledException的捕获，避免进程崩溃
给SocketIO客户端的异步操作单独添加异常处理，比如在连接、消息接收的回调中补充try-catch

2. 优化SocketIO连接配置

当前使用Polling传输协议，可调整超时参数：

this.Socket = new SocketIO(URL, new SocketIOOptions
{
    Query = new Dictionary<string, string>() { { "token", currentUser.TOKEN }, { "userID", currentUser.USERID }, { "apiType", "INTERACTIVE" } },
    EIO = 3,
    Transport = SocketIOClient.Transport.TransportProtocol.Polling,
    Path = "/interactive/socket.io",
    // 添加超时配置
    ConnectionTimeout = TimeSpan.FromSeconds(30),
    PollingTimeout = TimeSpan.FromSeconds(20)
});

3. 修复未等待异步操作的问题

StartSocket方法中调用_ = StartSocketAsync();未等待任务完成，可能导致异步操作中的异常无法被正确捕获。修改为：

public async Task StartSocket(User user, string URL)
{
    this.currentUser = user;
    this.URL = URL;
    await StartSocketAsync();
}

调用方需使用await调用StartSocket，确保异常能被上层捕获。

4. 完善重连逻辑

当前重连仅在StartSocketAsync的catch中触发，但TaskCanceledException可能未进入该catch块。可利用SocketIO的OnReconnectAttempt事件补充重连逻辑，或在全局异常处理中检测到TaskCanceledException时触发重连。

5. 日志增强

在Socket_OnDisconnected和OnError事件中，补充更详细的错误信息，包括异常类型和堆栈，便于定位是网络问题还是服务器端问题。

临时应急方案

如果无法立即修复，可添加全局异常处理器，捕获未处理的TaskCanceledException并记录日志，避免进程终止：

AppDomain.CurrentDomain.UnhandledException += (sender, e) =>
{
    if (e.ExceptionObject is TaskCanceledException tce)
    {
        FileLogger.logMessage("未处理的TaskCanceledException: " + tce.Message + tce.StackTrace);
        // 可选：触发重连逻辑
    }
};

内容的提问来源于stack exchange，提问作者Ankur Tripathi

Socket.IO连接随机触发未处理TaskCanceledException求助

错误信息 #

相关代码 #

核心原因定位 #

具体修复措施 #

1. 捕获并处理TaskCanceledException #

2. 优化SocketIO连接配置 #

3. 修复未等待异步操作的问题 #

4. 完善重连逻辑 #

5. 日志增强 #

临时应急方案 #