Aszinkron IEnumerable

A Task Parallel Library lehetőséget adott arra, hogy párhuzamosítsuk egy kollekció bejárását és feldolgozását. Az implementáció problémája az, hogy a feldolgozandó kollekció párhuzamosításának van egy előfeltétele: a kollekció elemeinek a memóriában kell lenniük.

Tételezzük fel, hogy webről töltünk le adatokat, amiken műveleteket kell végeznünk. Ebben az esetben a legcélszerűbb az, ha minden egyes letöltés egy külön Task, így a letöltések párhuzamosan tudnak futni. Azonban az adatok feldolgozását csak akkor tudjuk megkezdeni, ha a kollekció összes eleme letöltődött. Ez a mai párhuzamos adatfeldolgozás követelményeinek nem igen felel meg, mivel az adatok áramlása nem folyamatos.

A megoldást a C# 8.0-ban bemutatott, megtévesztő nevű Async Streams nyelvi szolgáltatás hidalja át. A név olyan szempontból megtévesztő, hogy első olvasatra hajlamos az ember a fájlkezelésnél bemutatott Stream osztályra asszociálni.

C# 8.0 újdonsága az IAsyncEnumerable<T> típus, ami aszinkron végrehajtást enged. Használatára nézzünk egy példát:

using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace PeldaAsyncStream
{
    class Program
    {
        //IAsyncEnumerable csak async módon deklarálható
        public static async IAsyncEnumerable<int> SzamSor()
        {
            for (int i = 0; i <= 20; i++)
            {
                await Task.Delay(200);
                yield return i;
            }
        }

        static async Task Main(string[] args)
        {
            //Aszinkron bejárás
            await foreach (var szam in SzamSor())
            {
                Console.Write("{0}, ", szam);
            }
            Console.ReadKey();
        }
    }
}

A program kimenete

0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20,

A program kimenetében statikusan szemlélve nincs semmi érdekes, azonban ha futtatjuk, akkor azt tapasztaljuk, hogy az első elem 200ms után jelenik meg a képernyőn, majd 200ms mulva a következő és így tovább. Amint végez a Task.Delay(200); hívás és a yield utasítás vérehajtódik, az elem feldolgozása azonnal megkezdődik, így folyamatos tud lenni az adatáramlás, még akkor is, ha egy elem feldolgozása hosszabb időt vesz igénybe.

Az IAsyncEnumerable típus érdekesége, hogy csak az async kulcsszóval deklarálható és elemeket csak yield return segítségével tud visszaadni, vagyis kollekciót nem tudunk visszaadni. Ez annak köszönhető, hogy az IAsyncEnumerable interfész nem implementálja az IEnumerable<T> interfészt.

Ez a technika is remekül használható termelő/fogyasztó jellegű problémákra, ahol a fogyasztó lazy módon húzza a termelőtől az újabb és újabb elemeket. Ennek a megoldásnak a termelő által tolt pipeline alternatívája lehet a Reactive Programming, de annak tárgyalása kívül esik jelen könyv keretein.