如何使用异步等待并行化此代码
How to parallelize this code using async await
我正在努力更好地理解异步编程。我编写了这段代码并添加了异步,但速度仍然相同。每个订阅调用都需要一秒钟以上,所以当我在控制台中打印它时,它仍在同步工作。
我只想异步执行每个订阅调用。另外因为第一个函数有很多订阅调用,我不确定是否应该异步或者我是否应该异步第二个调用。因为在程序开始时,我必须调用第一个函数 50 次。后来我只在添加新项目时调用第一个函数 1 次。
public static List<Task<CallResult<UpdateSubscription>>> AddNewSubscription(
BinanceSocketClient socketClient, Coin coin)
{
List<Task<CallResult<UpdateSubscription>>> tasks =
new List<Task<CallResult<UpdateSubscription>>>();
List<UpdateSubscription> subs = new List<UpdateSubscription>();
var sub = socketClient.Spot.SubscribeToSymbolTickerUpdatesAsync(coin.SymbolPair,
data =>
{
coin.Price = data.LastPrice;
coin.LowestDailyPrice = data.LowPrice;
coin.HighestDailyPrice = data.HighPrice;
coin.PriceChangeInPercentDaily = data.PriceChangePercent;
});
tasks.Add(sub);
subs.Add(sub.Result.Data);
sub = socketClient.Spot.SubscribeToKlineUpdatesAsync(coin.SymbolPair,
KlineInterval.OneHour, data =>
{
coin.LastHourOpenPrice = data.Data.Open;
coin.LastHourVolume = data.Data.BaseVolume;
coin.TotalBuyVolume = data.Data.TakerBuyBaseVolume;
coin.TotalSellVolume = coin.LastHourVolume - coin.TotalBuyVolume;
});
tasks.Add(sub);
subs.Add(sub.Result.Data);
sub = socketClient.Spot.SubscribeToKlineUpdatesAsync(coin.SymbolPair,
KlineInterval.OneMinute, data =>
{
coin.LastMinuteOpenPrice = data.Data.Open;
});
tasks.Add(sub);
subs.Add(sub.Result.Data);
Utility.ActiveSubscriptions.Add(coin.Symbol, subs);
return tasks;
}
public static async Task AddNewSubscriptions(BinanceSocketClient socketClient,
IEnumerable<Coin> coins)
{
List<Task<CallResult<UpdateSubscription>>> tasks =
new List<Task<CallResult<UpdateSubscription>>>();
int index = 0;
foreach (var coin in coins)
{
var subs = Utility.AddNewSubscription(socketClient, coin);
Console.WriteLine("{0}/{1} : {2}", index, coins.Count(), coin.Symbol);
++index;
tasks.Union(subs);
}
//foreach(var task in tasks)
// Utility.ActiveSubscriptions.Add ( coin.Symbol, task.Result.Data );
await Task.WhenAll(tasks);
}
public async static Task<List<CallResult<UpdateSubscription>>> AddNewSubscription ( BinanceSocketClient socketClient, IEnumerable<Coin> coins )
{
var tasks = new List<Task<CallResult<UpdateSubscription>>> ( );
var taskdict = new Dictionary<string, List<Task<CallResult<UpdateSubscription>>>> ( );
foreach ( var coin in coins )
{
var subtasks = Utility.CreateSubscriptionTasks ( socketClient, coin );
tasks.AddRange ( subtasks );
taskdict.Add ( coin.Symbol, subtasks );
}
await Task.WhenAll ( tasks );
foreach ( var task in taskdict )
{
var subs = task.Value.Select ( x => x.Result.Data ).ToList ( );
Utility.ActiveSubscriptions.Add ( task.Key, subs );
}
return tasks.Select ( x => x.Result ).ToList ( );
}
除非您想阻塞线程并等待结果,否则不要调用 .Result
。在这种情况下,您可能不想等到最后,在所有任务都已启动之后才得到结果。
public static List<Task<CallResult<UpdateSubscription>>> AddNewSubscription ( BinanceSocketClient socketClient, Coin coin )
{
var tasks = new List<Task<CallResult<UpdateSubscription>>> ( );
var sub = socketClient.Spot.SubscribeToSymbolTickerUpdatesAsync ( coin.SymbolPair, data =>
{
coin.Price = data.LastPrice;
coin.LowestDailyPrice = data.LowPrice;
coin.HighestDailyPrice = data.HighPrice;
coin.PriceChangeInPercentDaily = data.PriceChangePercent;
} );
tasks.Add ( sub );
sub = socketClient.Spot.SubscribeToKlineUpdatesAsync ( coin.SymbolPair, KlineInterval.OneHour, data =>
{
coin.LastHourOpenPrice = data.Data.Open;
coin.LastHourVolume = data.Data.BaseVolume;
coin.TotalBuyVolume = data.Data.TakerBuyBaseVolume;
coin.TotalSellVolume = coin.LastHourVolume - coin.TotalBuyVolume;
} );
tasks.Add ( sub );
sub = socketClient.Spot.SubscribeToKlineUpdatesAsync ( coin.SymbolPair, KlineInterval.OneMinute, data =>
{
coin.LastMinuteOpenPrice = data.Data.Open;
} );
tasks.Add ( sub );
//Here is where we wait for them all to complete in parallel
Task.WaitAll( tasks );
var subs = tasks.Select( x => x.Result.Data ).ToList();
Utility.ActiveSubscriptions.Add ( coin.Symbol, subs );
return tasks;
}
我正在努力更好地理解异步编程。我编写了这段代码并添加了异步,但速度仍然相同。每个订阅调用都需要一秒钟以上,所以当我在控制台中打印它时,它仍在同步工作。
我只想异步执行每个订阅调用。另外因为第一个函数有很多订阅调用,我不确定是否应该异步或者我是否应该异步第二个调用。因为在程序开始时,我必须调用第一个函数 50 次。后来我只在添加新项目时调用第一个函数 1 次。
public static List<Task<CallResult<UpdateSubscription>>> AddNewSubscription(
BinanceSocketClient socketClient, Coin coin)
{
List<Task<CallResult<UpdateSubscription>>> tasks =
new List<Task<CallResult<UpdateSubscription>>>();
List<UpdateSubscription> subs = new List<UpdateSubscription>();
var sub = socketClient.Spot.SubscribeToSymbolTickerUpdatesAsync(coin.SymbolPair,
data =>
{
coin.Price = data.LastPrice;
coin.LowestDailyPrice = data.LowPrice;
coin.HighestDailyPrice = data.HighPrice;
coin.PriceChangeInPercentDaily = data.PriceChangePercent;
});
tasks.Add(sub);
subs.Add(sub.Result.Data);
sub = socketClient.Spot.SubscribeToKlineUpdatesAsync(coin.SymbolPair,
KlineInterval.OneHour, data =>
{
coin.LastHourOpenPrice = data.Data.Open;
coin.LastHourVolume = data.Data.BaseVolume;
coin.TotalBuyVolume = data.Data.TakerBuyBaseVolume;
coin.TotalSellVolume = coin.LastHourVolume - coin.TotalBuyVolume;
});
tasks.Add(sub);
subs.Add(sub.Result.Data);
sub = socketClient.Spot.SubscribeToKlineUpdatesAsync(coin.SymbolPair,
KlineInterval.OneMinute, data =>
{
coin.LastMinuteOpenPrice = data.Data.Open;
});
tasks.Add(sub);
subs.Add(sub.Result.Data);
Utility.ActiveSubscriptions.Add(coin.Symbol, subs);
return tasks;
}
public static async Task AddNewSubscriptions(BinanceSocketClient socketClient,
IEnumerable<Coin> coins)
{
List<Task<CallResult<UpdateSubscription>>> tasks =
new List<Task<CallResult<UpdateSubscription>>>();
int index = 0;
foreach (var coin in coins)
{
var subs = Utility.AddNewSubscription(socketClient, coin);
Console.WriteLine("{0}/{1} : {2}", index, coins.Count(), coin.Symbol);
++index;
tasks.Union(subs);
}
//foreach(var task in tasks)
// Utility.ActiveSubscriptions.Add ( coin.Symbol, task.Result.Data );
await Task.WhenAll(tasks);
}
public async static Task<List<CallResult<UpdateSubscription>>> AddNewSubscription ( BinanceSocketClient socketClient, IEnumerable<Coin> coins )
{
var tasks = new List<Task<CallResult<UpdateSubscription>>> ( );
var taskdict = new Dictionary<string, List<Task<CallResult<UpdateSubscription>>>> ( );
foreach ( var coin in coins )
{
var subtasks = Utility.CreateSubscriptionTasks ( socketClient, coin );
tasks.AddRange ( subtasks );
taskdict.Add ( coin.Symbol, subtasks );
}
await Task.WhenAll ( tasks );
foreach ( var task in taskdict )
{
var subs = task.Value.Select ( x => x.Result.Data ).ToList ( );
Utility.ActiveSubscriptions.Add ( task.Key, subs );
}
return tasks.Select ( x => x.Result ).ToList ( );
}
除非您想阻塞线程并等待结果,否则不要调用 .Result
。在这种情况下,您可能不想等到最后,在所有任务都已启动之后才得到结果。
public static List<Task<CallResult<UpdateSubscription>>> AddNewSubscription ( BinanceSocketClient socketClient, Coin coin )
{
var tasks = new List<Task<CallResult<UpdateSubscription>>> ( );
var sub = socketClient.Spot.SubscribeToSymbolTickerUpdatesAsync ( coin.SymbolPair, data =>
{
coin.Price = data.LastPrice;
coin.LowestDailyPrice = data.LowPrice;
coin.HighestDailyPrice = data.HighPrice;
coin.PriceChangeInPercentDaily = data.PriceChangePercent;
} );
tasks.Add ( sub );
sub = socketClient.Spot.SubscribeToKlineUpdatesAsync ( coin.SymbolPair, KlineInterval.OneHour, data =>
{
coin.LastHourOpenPrice = data.Data.Open;
coin.LastHourVolume = data.Data.BaseVolume;
coin.TotalBuyVolume = data.Data.TakerBuyBaseVolume;
coin.TotalSellVolume = coin.LastHourVolume - coin.TotalBuyVolume;
} );
tasks.Add ( sub );
sub = socketClient.Spot.SubscribeToKlineUpdatesAsync ( coin.SymbolPair, KlineInterval.OneMinute, data =>
{
coin.LastMinuteOpenPrice = data.Data.Open;
} );
tasks.Add ( sub );
//Here is where we wait for them all to complete in parallel
Task.WaitAll( tasks );
var subs = tasks.Select( x => x.Result.Data ).ToList();
Utility.ActiveSubscriptions.Add ( coin.Symbol, subs );
return tasks;
}