异步流使用注意事项

Posted 2021-06-30 dotNET跨平台

异步流是C#8.0出来的特性，可以让流式数据以尽快分批的方式返回，而不像之前那样，当全部获取到数据后，再返回。

下面的例子的场景有点特殊，查询一个较多的数据，为了查询不给数据带来很大压力，做成一个分批从数据查询回数据，组装成一个整理List返回。

如果是之前的做法，就需要等所有的数据返回后，累加到一个集合中，然后返回；如果用异步流，可以返回一批，就返回，以便更快的让使用者展示或使用。

异步流也有问题，多次返回，虽然用户体验相对好一些，但用时更长，通过下面的例子可以了解到。

using System;
using System.Collections.Generic;
using System.Data.SqlClient;
using System.Linq;
using System.Net.WebSockets;
using System.Text;
using System.Threading.Tasks;
using System.Diagnostics;
using System.Security.Cryptography;
using System.IO;
using Dapper;


namespace KeyWordsDemo
{
    class AsyncStreamDemo : IDemoAsync
    {
        public async Task RunAsync()
        {
            Console.WriteLine("1、异常流  2、同步获取")
            var no = Console.ReadLine();
            var producer = new Producer();
            if(no == "1")
            {
                var watch = new Stopwatch();
                watch.Start();


                await foreach (var orders in producer.EnumerateOrdersAsync())
                {
                    foreach (var order in orders)
                    {
                        Console.WriteLine(order.ToString());
                    }
                }
                watch.Stop();
                Console.Title = $"时长：{ watch.Elapsed.TotalSeconds}";
            }
            else
            {
                var watch = new Stopwatch();
                watch.Start();
                var orders = await producer.GetOrdersAsync();
                foreach (var order in orders)
                {
                    Console.WriteLine(order.ToString());
                }
                watch.Stop();
                Console.Title = $"获取到数据：{orders.Count()} 时长：{ watch.Elapsed.TotalSeconds}";
            }
            Console.ReadLine();
        }
    }
    class Producer
    {
        public async Task<List<SalesOrderDetail>> GetOrdersAsync()
        {
            var orders = new List<SalesOrderDetail>();
            var offset = 0;
            while (true)
            {
                var list = (await QueryOrdersAsync(offset)).ToList();
                orders.AddRange(list);
                offset++;
                if (list.Count < 100)
                {
                    break;
                }
            }
            return orders;
        }


        public async IAsyncEnumerable<List<SalesOrderDetail>> EnumerateOrdersAsync()
        {
            var offset = 0;
            while (true)
            {
                var list = (await QueryOrdersAsync(offset)).ToList();
                yield return list;
                offset++;
                if (list.Count < 100)
                {
                    break;
                }
            }
        }
        public async Task<IEnumerable<SalesOrderDetail>> QueryOrdersAsync(int offset)
        {
            using var con = new SqlConnection("server=.;database=AdventureWorks2016;uid=sa;pwd=sa;");
            var sql = @$"select * from Sales.SalesOrderDetail order by SalesOrderID,SalesOrderDetailID  offset {offset * 100} row fetch next 100 row only";
            return await con.QueryAsync<SalesOrderDetail>(sql);
        }
    }
    class SalesOrderDetail
    {
        public int SalesOrderID { get; set; }
        public string CarrierTrackingNumber { get; set; }
        public short OrderQty { get; set; }
        public int ProductID { get; set; }
        public int SpecialOfferID { get; set; }
        public decimal UnitPrice { get; set; }
        public decimal UnitPriceDiscount { get; set; }
        public Guid rowguid { get; set; }
        public DateTime ModifiedDate { get; set; }
        public override string ToString()
        {
            return System.Text.Json.JsonSerializer.Serialize(this);
        }
    }
}