并行异步操作，超时后如何"kill"一个？

Question

我有一个程序，用于通过SNMP从我的网络获取一些数据。我使用最后的.NET和SharpSnmp库。我意识到一个 ForEachAsync 方法，就像这篇文章中暴露的那样。因此，我可以并行执行 snmp 请求（并详细说明响应），为列表中的每个设备创建一个任务。它可以工作，但如果设备由于某种原因没有回复，我的程序就会卡住。所以我需要管理某种超时来"杀死"库公开的异步函数。这就是我在foreachAsync中调用的函数：

 public static async Task<Tuple<string, List<Variable>, Exception>>
            GetAsync(string ip, IEnumerable<string> vars, int timeout = 5000)
        {
            try
            {
                IPAddress agentIp;
                bool parsed = IPAddress.TryParse(ip, out agentIp);
                if (!parsed)
                {
                    foreach (IPAddress address in
                        Dns.GetHostAddresses(ip).Where(address => address.AddressFamily == AddressFamily.InterNetwork))
                    {
                        agentIp = address;
                        break;
                    }
                    if (agentIp == null)
                        throw new Exception("Impossibile inizializzare la classe CGesSnmp senza un indirizzo IP valido");
                }
                IPEndPoint receiver = new IPEndPoint(agentIp, LOCAL_PORT);
                VersionCode version = VersionCode.V2;
                string community = "public";
                List<Variable> vList = new List<Variable>();
                foreach (string s in vars)
                    vList.Add(new Variable(new ObjectIdentifier(s)));
                 // This is the function I want to "stop" or "kill" in some way
                List<Variable> result = (List<Variable>)await Messenger.GetAsync(version, receiver, new OctetString(community), vList);
                return new Tuple<string, List<Variable>, Exception>(ip, result, null);
            }
            catch (Exception ex)
            {
                return new Tuple<string, List<Variable>, Exception>(ip, null, ex);
            }
        }  

Answer 1

if 方法支持TaskCancellationToken

public static async Task<Tuple<string, List<Variable>, Exception>>
            GetAsync(string ip, IEnumerable<string> vars, int timeout = 5000)
        {
            try
            {
                IPAddress agentIp;
                bool parsed = IPAddress.TryParse(ip, out agentIp);
                if (!parsed)
                {
                    foreach (IPAddress address in
                        Dns.GetHostAddresses(ip).Where(address => address.AddressFamily == AddressFamily.InterNetwork))
                    {
                        agentIp = address;
                        break;
                    }
                    if (agentIp == null)
                        throw new Exception("Impossibile inizializzare la classe CGesSnmp senza un indirizzo IP valido");
                }
                IPEndPoint receiver = new IPEndPoint(agentIp, LOCAL_PORT);
                VersionCode version = VersionCode.V2;
                string community = "public";
                List<Variable> vList = new List<Variable>();
                foreach (string s in vars)
                    vList.Add(new Variable(new ObjectIdentifier(s)));
                 // This is the function I want to "stop" or "kill" in some way
                CancellationTokenSource cancel = new CancellationTokenSource(TimeSpan.FromSeconds(timeout));
                List<Variable> result = (List<Variable>)await Messenger.GetAsync(version, receiver, new OctetString(community), vList, cancel.Token);
                return new Tuple<string, List<Variable>, Exception>(ip, result, null);
            }
            catch (Exception ex)
            {
                return new Tuple<string, List<Variable>, Exception>(ip, null, ex);
            }
        }  

否则

public static Task<Tuple<string, List<Variable>, Exception>>
            GetAsync(string ip, IEnumerable<string> vars, int timeout = 5000){
 TaskCompletionSource<Tuple<string,List<Variable>>> taskSource = 
       new TaskCompletionSource<Tuple<string,List<Variable>>>();
 Task.Run(async ()=> {
     var result = await GetAsync(ip,vars);
     taskSource.TrySetResult(result);
 });
 Task.Run(async ()=>{
     await Task.Delay(TimeSpan.FromSeconds(timeout));
     taskSource.TrySetCancelled();
 });
 return taskSource.Task;
}

private static async Task<Tuple<string, List<Variable>, Exception>>
            _GetAsync(string ip, IEnumerable<string> vars)
        {
            try
            {
                IPAddress agentIp;
                bool parsed = IPAddress.TryParse(ip, out agentIp);
                if (!parsed)
                {
                    foreach (IPAddress address in
                        Dns.GetHostAddresses(ip).Where(address => address.AddressFamily == AddressFamily.InterNetwork))
                    {
                        agentIp = address;
                        break;
                    }
                    if (agentIp == null)
                        throw new Exception("Impossibile inizializzare la classe CGesSnmp senza un indirizzo IP valido");
                }
                IPEndPoint receiver = new IPEndPoint(agentIp, LOCAL_PORT);
                VersionCode version = VersionCode.V2;
                string community = "public";
                List<Variable> vList = new List<Variable>();
                foreach (string s in vars)
                    vList.Add(new Variable(new ObjectIdentifier(s)));
                 // This is the function I want to "stop" or "kill" in some way
                List<Variable> result = (List<Variable>)await Messenger.GetAsync(version, receiver, new OctetString(community), vList);
                return new Tuple<string, List<Variable>, Exception>(ip, result, null);
            }
            catch (Exception ex)
            {
                return new Tuple<string, List<Variable>, Exception>(ip, null, ex);
            }
        }  

Answer 2

Task<[whatever type GetAsync returns]> messengerTask = Messenger.GetAsync(version, receiver, new OctetString(community), vList);

if (await Task.WhenAny(messengerTask, Task.Delay(timeout)) == messengerTask)
{
//GetAsync completed within timeout. Access task's result using
// messengerTask.Result. You can also check if the task RanToCompletion.
} 
else
{ 
//timeout    
}