如何使用 async await 并行化此代码

如何解决如何使用 async await 并行化此代码

我试图更好地理解异步编程。我写了这段代码添加了 async 但速度仍然相同。每个订阅调用都需要一秒钟以上,所以当我在控制台中打印它时,它仍然在同步工作。

我想要的是异步执行每个订阅调用。也因为第一个函数有很多订阅调用,我不确定它是否应该是异步的,或者我是否应该异步第二个调用。因为在程序开始时,我必须调用一个函数 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;
}

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐


Selenium Web驱动程序和Java。元素在(x,y)点处不可单击。其他元素将获得点击?
Python-如何使用点“。” 访问字典成员?
Java 字符串是不可变的。到底是什么意思?
Java中的“ final”关键字如何工作?(我仍然可以修改对象。)
“loop:”在Java代码中。这是什么,为什么要编译?
java.lang.ClassNotFoundException:sun.jdbc.odbc.JdbcOdbcDriver发生异常。为什么?
这是用Java进行XML解析的最佳库。
Java的PriorityQueue的内置迭代器不会以任何特定顺序遍历数据结构。为什么?
如何在Java中聆听按键时移动图像。
Java“Program to an interface”。这是什么意思?
Java在半透明框架/面板/组件上重新绘画。
Java“ Class.forName()”和“ Class.forName()。newInstance()”之间有什么区别?
在此环境中不提供编译器。也许是在JRE而不是JDK上运行?
Java用相同的方法在一个类中实现两个接口。哪种接口方法被覆盖?
Java 什么是Runtime.getRuntime()。totalMemory()和freeMemory()?
java.library.path中的java.lang.UnsatisfiedLinkError否*****。dll
JavaFX“位置是必需的。” 即使在同一包装中
Java 导入两个具有相同名称的类。怎么处理?
Java 是否应该在HttpServletResponse.getOutputStream()/。getWriter()上调用.close()?
Java RegEx元字符(。)和普通点?