主页 / 问题 / 762610
Accepted
Monk
Asked:2020-12-24 18:46:43 +0000 UTC2020-12-24 18:46:43 +0000 UTC

如何使异步 IEnumerable？

  • 772

我正在同时在几个地方进行搜索。

目前，在顶层，类似：

  foreach (var item in Plugins.SelectMany(p => p.Search(query)))
    Items.Add(new ViewModel(item));

在 Search 方法的每个实现中：

public override IEnumerable<Item> Search(string name)
{
  await some resource
  foreach (var element in networkResult)
    ...
    yield return result;
}

事实上，我希望并行访问所有搜索，以便每个元素在准备就绪时出现在 UI 中，而不是在整个事情结束时出现，因为它目前可以工作。究竟要在这里包含什么任务 - 没有好主意。在外面，它似乎更合乎逻辑Task<IEnumerable>，但我不明白如何正确实现它。

c#

1 个回答

  1. Best Answer

    看，没那么难。

    您安装 Ix、nuget 包System.Interactive和System.Interactive.Async. 你有一个接口IAsyncEnumerable和帮助类。

    你可以这样使用：

    static class FileEx
{
    public static IAsyncEnumerable<string> ReadLinesAsync(string path)
    {
        // IAsyncEnumerable<T> обладает только одной функцией - создать энумератор
        return AsyncEnumerable.CreateEnumerable(() =>
        {
            var stream = File.OpenText(path);
            string current = null;
            // создаём энумератор при помощи готовой фабрики
            return AsyncEnumerable.CreateEnumerator(
                // у StreamReader.ReadLineAsync нет перегрузки с ct, пичалько
                moveNext: async ct => (current = await stream.ReadLineAsync()) != null,
                current: () => current,
                dispose: stream.Dispose);
        });
    }
}

    看看这里发生了什么。对于初学者来说，相同的序列可以通过穿插的不同代码段运行，因此我们将当前旁路的状态保留在枚举器中。原则上，我们需要为枚举器创建一个单独的类，并在其中保留属性。但我们将采用更时尚的方式，并将数据保存在闭包中。我们打开StreamReader，为当前行设置一个变量。

    异步MoveNext迭代器函数从 'a 获取下一行StreamReader，并检查结果是否为null(null表示文件结尾)。该函数Current只返回当前行。该函数Dispose在最后关闭流。

    现在你可以使用这个：

    class Program
{
    static async Task Main(string[] args)
    {
        var lines = FileEx.ReadLinesAsync("text.txt");
        using (var en = lines.GetEnumerator())
        {
            while (await en.MoveNext())
                Console.WriteLine(en.Current);
        }
    }
}

    或者干脆

    await FileEx.ReadLinesAsync("text.txt")
            .ForEachAsync(s => Console.WriteLine(s));

    下一版本的 C# 计划直接在该语言中支持异步枚举器。有了它，我们的例子会写成这样：

    static class FileEx
{
    public static async IAsyncEnumerable<string> ReadLinesAsync(string path)
    {
        using (var stream = File.OpenText(path))
        {
            string current;
            while ((current = await stream.ReadLineAsync()) != null)
                yield return current;
        };
    }
}

class Program
{
    static async Task Main(string[] args)
    {
        foreach await (var s in FileEx.ReadLinesAsync("text.txt"))
            Console.WriteLine(s);
    }
}

    看，对于您的特定情况（这里是这样的代码），您应该将任务分解为其组成部分，因为其中有很多异步的东西。然后，您可以改为链接它们。

    让我们从获取页面并解析它们开始。主机列表很容易获取，不需要异步：

    var hosts = ConfigStorage.Plugins
    .Where(p => p.GetParser().GetType() == typeof(Parser))
    .Select(p => p.GetSettings().MainUri);

    现在，我们需要通过 host 获取列表HtmlNodeCollection。这是一个“长”任务，我们把它放在一个任务中：

    async Task<HtmlNodeCollection> GetHostMangasAsync(string name, Uri host, CookieClient client)
{
    var searchHost = new Uri(host, "search?q=" + WebUtility.UrlEncode(name));
    var page = await Task.Run(() => Page.GetPage(searchHost, client));
    if (!page.HasContent)
        return null;

    return await Task.Run(() =>
    {
        var document = new HtmlDocument();
        document.LoadHtml(page.Content);
        return document.DocumentNode.SelectNodes("//div[@class='tile col-sm-6']");
    });
}

    现在，我们需要从主机的非异步序列和Task'a中获取异步序列，它从每个主机接收HtmlNodeCollection。开箱即用的Ix中我没有找到这样的方法，但是很容易自己组合起来。让我们把它概括一下，以防您需要更多。该代码实际上与带有File.ReadLinesAsync.

    static class AsyncEnumerableExtensions
{
    public static IAsyncEnumerable<R> SelectAsync<T, R>(
        this IEnumerable<T> seq, Func<T, Task<R>> selector)
    {
        return AsyncEnumerable.CreateEnumerable(() =>
        {
            IEnumerator<T> seqEnum = seq.GetEnumerator();
            R current = default;
            return AsyncEnumerable.CreateEnumerator(
                moveNext: async ct =>
                {
                    if (!seqEnum.MoveNext())
                        return false;
                    current = await selector(seqEnum.Current);
                    return true;
                },
                current: () => current,
                dispose: seqEnum.Dispose);
        });
    }
}

    有了这个，我们可以这样写：

    IAsyncEnumerable<HtmlNodeCollection> GetSearchPages(string name)
{
    var hosts = ConfigStorage.Plugins
        .Where(p => p.GetParser().GetType() == typeof(Parser))
        .Select(p => p.GetSettings().MainUri);
    var client = new CookieClient();
    return hosts.SelectAsync(host => GetHostMangasAsync(name, host, client)))
                .Where(nc => nc != null);
}

    需要检查，null因为它GetHostMangasAsync可以返回null。

    太好了，让我们继续。所以，我们再次有一个非异步集合HtmlNodeCollection，我们可以使用异步函数从其中的每个元素中提取一个实例（因为我们有网络访问权限）IManga。我们编写代码：

    async Task<IManga> GetMangaFromNode(Uri host, CookieClient client, HtmlNode manga)
{
    // Это переводчик, идем дальше.
    if (manga.SelectSingleNode(".//i[@class='fa fa-user text-info']") != null)
        return null;

    var image = manga.SelectSingleNode(".//div[@class='img']//a//img");
    var imageUri = image?.Attributes.Single(a => a.Name == "data-original").Value;

    var mangaNode = manga.SelectSingleNode(".//h3//a");
    var mangaUri = mangaNode.Attributes.Single(a => a.Name == "href").Value;
    var mangaName = mangaNode.Attributes.Single(a => a.Name == "title").Value;

    if (!Uri.TryCreate(mangaUri, UriKind.Relative, out Uri test))
        return null;

    var result = Mangas.Create(new Uri(host, mangaUri));
    result.Name = WebUtility.HtmlDecode(mangaName);
    if (imageUri != null)
        result.Cover = await client.DownloadDataAsync(imageUri);
    return result;
}

    我们现在需要连接它们。这并不难。唯一的问题是GetMangaFromNode你也需要它CookieClient，但我们把它藏在里面GetSearchPages。好吧，让我们把它传到外面。然后，我们GetSearchPages只返回 from HtmlNodeCollection，但我们还需要host。我们修改GetSearchPages：我们将从主机和集合返回对HtmlNode，并作为输入CookieClient：

    IAsyncEnumerable<(Uri host, HtmlNodeCollection nodes)> GetSearchPages(
        string name, CookieClient client)
{
    var hosts = ConfigStorage.Plugins
        .Where(p => p.GetParser().GetType() == typeof(Parser))
        .Select(p => p.GetSettings().MainUri);
    return
        hosts.SelectAsync(
                async host => (host, nodes: await GetHostMangasAsync(name, host, client)))
             .Where(pair => pair.nodes != null);
}

    好吧，让我们结合起来。我们为每个同步集合HtmlNode使用了一个异步函数，给出了一个集合实例IManga。这再次通过我们的SelectAsync:

    IAsyncEnumerable<IManga> GetFromHostAndNodes(
        Uri host, HtmlNodeCollection nodes, CookieClient client) =>
    nodes.SelectAsync(node => GetMangaFromNode(host, client, node));

    现在你可以把拼图放在一起：

    public IAsyncEnumerable<IManga> Search(string name)
{
    var client = new CookieClient();
    return
        GetSearchPages(name, client)
            .SelectMany(pair => GetFromHostAndNodes(pair.host, pair.nodes, client))
            .Where(m => m != null);
}

    一切！

    • 9

相关问题