Orleans rozhraní API streamování

Aplikace komunikují se streamy prostřednictvím rozhraní API velmi podobně jako dobře známá reaktivní rozšíření (Rx) v .NET. Hlavním rozdílem je, že rozšíření datových proudů jsou Orleans, aby bylo zpracování ve distribuovaných a škálovatelných výpočetních prostředcích infrastruktury efektivnější.

Asynchronní stream

Začnete tím, že pomocí zprostředkovatele streamu získáte identifikátor datového proudu. Poskytovatele datových proudů si můžete představit jako objekt pro vytváření datových proudů, který implementátorům umožňuje přizpůsobit chování datových proudů a sémantiku:

public async Task SetupStream()
{
    IStreamProvider streamProvider = this.GetStreamProvider("SimpleStreamProvider");
    StreamId streamId = StreamId.Create("MyStreamNamespace", this.GetPrimaryKey());
    IAsyncStream<string> stream = streamProvider.GetStream<string>(streamId);
}

IStreamProvider streamProvider = base.GetStreamProvider("SimpleStreamProvider");
IAsyncStream<T> stream = streamProvider.GetStream<T>(Guid, "MyStreamNamespace");

Odkaz na zprostředkovatele datového proudu můžete získat buď voláním metody Grain.GetStreamProvider uvnitř zrna, nebo voláním metody GetStreamProvider na instanci klienta.

Orleans.Streams.IAsyncStream<T> je logický, silně typovaný popisovač virtuálního proudu, podobně jako Orleans Grain Reference. Volání a GetStreamProviderGetStream jsou čistě místní. Argumenty pro GetStream jsou GUID a další řetězec nazývaný obor streamu (který může být null). Identifikátor GUID a řetězec oboru názvů společně tvoří identitu datového proudu (podobně jako argumenty pro IGrainFactory.GetGrain). Tato kombinace poskytuje větší flexibilitu při určování identit datových proudů. Stejně jako grain 7 může existovat v rámci PlayerGrain typu a jiný grain 7 může existovat v rámci ChatRoomGrain typu, může stream 123 existovat v rámci PlayerEventsStream oboru názvů a jiný stream 123 může existovat v rámci ChatRoomMessagesStream oboru názvů.

Výroba a spotřeba

IAsyncStream<T> implementuje rozhraní IAsyncObserver<T> i IAsyncObservable<T> rozhraní. To vaší aplikaci umožňuje používat stream buď k vytváření nových událostí pomocí IAsyncObserver<T> , nebo k přihlášení k odběru a využívání událostí pomocí IAsyncObservable<T>.

public interface IAsyncObserver<in T>
{
    Task OnNextAsync(T item, StreamSequenceToken token = null);
    Task OnCompletedAsync();
    Task OnErrorAsync(Exception ex);
}

public interface IAsyncObservable<T>
{
    Task<StreamSubscriptionHandle<T>> SubscribeAsync(IAsyncObserver<T> observer);
}

Aby vaše aplikace vytvářela události do datového proudu, volá:

await stream.OnNextAsync<T>(event)

Pokud se chcete přihlásit k odběru datového proudu, vaše aplikace volá:

StreamSubscriptionHandle<T> subscriptionHandle = await stream.SubscribeAsync(IAsyncObserver)

SubscribeAsync Argumentem může být buď objekt implementující IAsyncObserver<T> rozhraní, nebo kombinace funkcí lambda pro zpracování příchozích událostí. Další možnosti SubscribeAsync jsou k dispozici prostřednictvím AsyncObservableExtensions třídy. SubscribeAsync vrátí StreamSubscriptionHandle<T>, neprůhledný popisovač použitý k odhlášení odběru ze streamu (podobně jako asynchronní verze IDisposable).

await subscriptionHandle.UnsubscribeAsync()

Je důležité si uvědomit, že předplatné je určené pro zrno, ne pro aktivaci. Jakmile se kód komponenty přihlásí k odběru datového proudu, toto předplatné překročí životnost této aktivace a zůstane trvalé, dokud se kód komponenty (potenciálně v jiné aktivaci) explicitně neodhlásí. Toto je jádro abstrakce virtuálního datového proudu: nejen, že všechny streamy vždy existují logicky, ale odběr streamu je také trvalý a žije mimo konkrétní fyzickou aktivaci, která ho vytvořila.

Násobnost

Datový Orleans proud může mít více výrobců a více spotřebitelů. Zpráva publikovaná producentem se před publikováním zprávy doručí všem uživatelům, kteří se přihlásili k odběru datového proudu.

Spotřebitel se navíc může přihlásit k odběru stejného datového proudu vícekrát. Pokaždé, když se přihlásí k odběru, dostane zpět jedinečný StreamSubscriptionHandle<T>. Pokud se obilí (nebo klient) přihlásí k odběru Xkrát stejného streamu, obdrží stejnou událost Xkrát, vždy pro každé předplatné. Příjemce se také může odhlásit od odběru individuálního předplatného. Všechna aktuální předplatná najdete voláním:

IList<StreamSubscriptionHandle<T>> allMyHandles =
    await IAsyncStream<T>.GetAllSubscriptionHandles();

Zotavení ze selhání

Pokud producent streamu zemře (nebo se deaktivuje jeho zrno), nemusí nic dělat. Až tato grain příště bude chtít vytvořit více událostí, může znovu získat popisovač proudu a vytvořit nové události jako obvykle.

Spotřebitelská logika je o něco více zapojená. Jak už bylo zmíněno dříve, jakmile se zrno přihlásí k odběru datového proudu, je toto předplatné platné, dokud se zrno explicitně neodhlásí. Pokud příjemce datového proudu zemře (nebo se deaktivuje jeho agregační interval) a vygeneruje se v datovém proudu nová událost, konzumentská agregace se automaticky znovu aktivuje (stejně jako jakákoli běžná Orleans agregace se automaticky aktivuje při odeslání zprávy). Jediné, co teď potřebuje kód zrnitosti, je poskytnout IAsyncObserver<T> pro zpracování dat. Příjemce musí v rámci metody znovu připojit logiku OnActivateAsync() zpracování. K tomu může volat:

StreamSubscriptionHandle<int> newHandle =
    await subscriptionHandle.ResumeAsync(IAsyncObserver);

Uživatel použije předchozí popisovač získaný během počátečního předplatného k obnovení zpracování. Všimněte si, že ResumeAsync pouze aktualizuje stávající předplatné novou instancí IAsyncObserver<T> logiky a nemění skutečnost, že tento uživatel je již přihlášen k odběru tohoto streamu.

Jak spotřebitel získá starou subscriptionHandle? Existují dvě možnosti. Příjemce mohl zachovat popisovač vrácený z původní SubscribeAsync operace a nyní ho může použít. Pokud spotřebitel nemá identifikátor, může IAsyncStream<T> přímá požádat o všechny aktivní identifikátory svých předplatných voláním:

IList<StreamSubscriptionHandle<T>> allMyHandles =
    await IAsyncStream<T>.GetAllSubscriptionHandles();

Příjemce pak může obnovit všechny nebo odhlásit odběr některých, pokud je to žádoucí.

Návod

Pokud konzumentské zrnko implementuje IAsyncObserver<T> rozhraní přímo (public class MyGrain<T> : Grain, IAsyncObserver<T>), neměl by teoreticky muset znovu připojit IAsyncObserver<T> , a proto by nemusel volat ResumeAsync. Streamovací runtime by měl automaticky zjistit, že IAsyncObserver<T> je již implementováno, a vyvolá tyto IAsyncObserver<T> metody. Streamovací runtime to ale v současné době nepodporuje a kód zrnka musí explicitně volat ResumeAsync, i když zrnko přímo implementuje IAsyncObserver<T>.

Explicitní a implicitní předplatná

Ve výchozím nastavení se musí příjemce proudu explicitně přihlásit k odběru datového proudu. Tento odběr je obvykle aktivován externí zprávou, kterou obdrží zrno (nebo klient) s pokynem k registraci. Například, když se uživatel připojí k některé chatovací místnosti, chatovací služba mu odešle zprávu JoinChatGroup s názvem chatu, což způsobí, že se uživatel přihlásí k odběru tohoto chatového streamu.

Streamy navíc Orleans podporují implicitní předplatná. V tomto modelu se zrno explicitně nepřihlašuje. Odebírá se automaticky a implicitně na základě jeho odstupňované identity a objektu ImplicitStreamSubscriptionAttribute. Hlavní hodnota implicitních předplatných spočívá v tom, že aktivita streamů umožňuje automaticky spustit aktivaci zrna (a tím i předplatného). Pokud by například jedno zrno chtělo vytvořit SMS proud a jiné zrno jej zpracovat, producent by potřeboval identitu spotřebitelského zrna a musel by ho vyzvat, aby se přihlásilo k odběru. Teprve potom může začít odesílat události. Místo toho může producent s implicitními předplatnými začít vytvářet události do datového proudu a spotřebitelský grain se automaticky aktivuje a přihlásí k odběru. V tomto případě producent nemusí vědět, kdo události čte.

Implementace zrnitosti MyGrainType může deklarovat atribut [ImplicitStreamSubscription("MyStreamNamespace")]. To říká modulu runtime streamování, že když je událost vygenerována ve streamu s identifikátorem GUID XXX a oborem názvů "MyStreamNamespace", měla by být doručena do grainu s identitou XXX typu MyGrainType. To znamená, že modul runtime mapuje datový proud <XXX, MyStreamNamespace> na agregační interval <XXX, MyGrainType>příjemce .

Přítomnost ImplicitStreamSubscription způsobí, že modul runtime streamování se automaticky přihlásí k odběru streamu a doručí mu události streamu. Kód jádra ale stále potřebuje sdělit běhové prostředí, jak má zpracovávat události. V podstatě musí připojit IAsyncObserver<T>. Proto při aktivaci zrna musí kód uvnitř OnActivateAsync volat:

public override async Task OnActivateAsync(CancellationToken cancellationToken)
{
    IStreamProvider streamProvider =
        this.GetStreamProvider("SimpleStreamProvider");

    StreamId streamId =
        StreamId.Create("MyStreamNamespace", this.GetPrimaryKey());
    IAsyncStream<string> stream =
        streamProvider.GetStream<string>(streamId);

    StreamSubscriptionHandle<string> subscription =
        await stream.SubscribeAsync(new MyStreamObserver());
}

IStreamProvider streamProvider =
    base.GetStreamProvider("SimpleStreamProvider");

IAsyncStream<T> stream =
    streamProvider.GetStream<T>(this.GetPrimaryKey(), "MyStreamNamespace");

StreamSubscriptionHandle<T> subscription =
    await stream.SubscribeAsync(IAsyncObserver<T>);

Psaní logiky předplatného

Níže jsou uvedeny pokyny pro zápis logiky předplatného pro různé případy: explicitní a implicitní předplatná, převinutelná a nepřevinutelná streamy. Hlavním rozdílem mezi explicitními a implicitními předplatnými je to, že pro implicitní předplatná existuje vždy přesně jedno takové předplatné na každý obor názvů streamu. Neexistuje žádný způsob, jak vytvořit více předplatných (bez násobnosti předplatného), žádný způsob zrušení předplatného a logika pro zrno potřebuje pouze připojit logiku zpracování. To také znamená, že nikdy není potřeba obnovit implicitní předplatné. Na druhou stranu pro explicitní předplatná musíte obnovit předplatné; jinak opětovné přihlášení k odběru vede k vícenásobnému přihlášení k odběru.

Implicitní předplatná:

U implicitních předplatných se stále musí přihlásit k odběru, aby se připojila logika zpracování. Můžete to provést v zrnu uživatele implementací IStreamSubscriptionObserver a IAsyncObserver<T> rozhraní, což umožní aktivaci zrna odděleně od procesu přihlášení k odběru. Pokud se chcete přihlásit k odběru datového proudu, vytvoří agregační interval popisovač a volání await handle.ResumeAsync(this) ve své OnSubscribed(...) metodě.

Ke zpracování zpráv implementujte metodu IAsyncObserver<T>.OnNextAsync(...) pro příjem datového proudu a sekvenčního tokenu. Alternativně ResumeAsync může metoda převzít sadu delegátů představující metody IAsyncObserver<T> rozhraní: onNextAsync, onErrorAsynca onCompletedAsync.

public Task OnNextAsync(string item, StreamSequenceToken? token = null)
{
    _logger.LogInformation("Received an item from the stream: {Item}", item);
    return Task.CompletedTask;
}

public async Task OnSubscribed(IStreamSubscriptionHandleFactory handleFactory)
{
    var handle = handleFactory.Create<string>();
    await handle.ResumeAsync(this);
}

public override async Task OnActivateAsync()
{
    var streamProvider = this.GetStreamProvider(PROVIDER_NAME);
    var stream =
        streamProvider.GetStream<string>(
            this.GetPrimaryKey(), "MyStreamNamespace");

    await stream.SubscribeAsync(OnNextAsync);
}

Explicitní předplatná:

Pro explicitní předplatná musí volat SubscribeAsync odběr streamu. Tímto se vytvoří předplatné a připojí se k němu zpracovatelská logika. Explicitní předplatné existuje, dokud se neodhlásí grain. Pokud se zrno deaktivuje a znovu aktivuje, je stále explicitně přihlášeno k odběru, ale není k němu připojena žádná logika zpracování. V tomto případě je potřeba znovu připojit logiku zpracování. Chcete-li to provést, ve svém OnActivateAsync, zrno nejprve musí zjistit své předplatné voláním IAsyncStream<T>.GetAllSubscriptionHandles(). Zrno musí spustit ResumeAsync na každém popisovači, se kterým chce pokračovat ve zpracování, nebo UnsubscribeAsync na jakýchkoli popisovačích, se kterými je hotovo. Datové zrno může také volitelně specifikovat StreamSequenceToken jako argument volání ResumeAsync, což způsobí, že toto explicitní předplatné začne čerpat z daného tokenu.

public override async Task OnActivateAsync(CancellationToken cancellationToken)
{
    var streamProvider = this.GetStreamProvider(PROVIDER_NAME);
    var streamId = StreamId.Create("MyStreamNamespace", this.GetPrimaryKey());
    var stream = streamProvider.GetStream<string>(streamId);

    var subscriptionHandles = await stream.GetAllSubscriptionHandles();
    foreach (var handle in subscriptionHandles)
    {
        await handle.ResumeAsync(this);
    }
}

public async override Task OnActivateAsync()
{
    var streamProvider = this.GetStreamProvider(PROVIDER_NAME);
    var stream =
        streamProvider.GetStream<string>(this.GetPrimaryKey(), "MyStreamNamespace");

    var subscriptionHandles = await stream.GetAllSubscriptionHandles();
    if (!subscriptionHandles.IsNullOrEmpty())
    {
        subscriptionHandles.ForEach(
            async x => await x.ResumeAsync(OnNextAsync));
    }
}

Pořadí a sekvence tokenů streamu

Pořadí doručování událostí mezi individuálním producentem a příjemcem závisí na poskytovateli streamu.

Se zprávou SMS producent explicitně řídí pořadí událostí, které uživatel vidí, tím, že řídí, jak je publikuje. Ve výchozím nastavení (pokud je možnost poskytovatele SMS serveru SimpleMessageStreamProviderOptions.FireAndForgetDeliveryfalse) a pokud producent čeká na každé OnNextAsync volání, události přicházejí v pořadí FIFO. Ve zprávách SMS se producent rozhodne, jak zpracovat selhání doručení signalizované chybou Task vrácenou voláním OnNextAsync.

Streamy front Azure nezaručují pořadí FIFO, protože podkladové fronty Azure nezaručují pořadí v případech selhání (i když zaručují pořadí FIFO při provádění bez selhání). Když producent vytvoří událost do fronty Azure, pokud operace fronty selže, musí se producent pokusit o jinou frontu a později se vypořádat s potenciálními duplicitními zprávami. Na straně doručování běhové prostředí streamování událost odebere z fronty a pokusí se ji doručit k zpracování příjemcům. Modul runtime odstraní událost z fronty pouze po úspěšném zpracování. Pokud se doručení nebo zpracování nezdaří, událost se z fronty neodstraní a později se automaticky znovu zobrazí. Modul runtime streamování se ho pokusí znovu doručit a potenciálně přeruší objednávku FIFO. Toto chování odpovídá obvyklým vlastnostem sémantiky front Azure.

Pořadí definované aplikací: Pokud chcete vyřešit výše uvedené problémy s řazením, může vaše aplikace volitelně zadat řazení. Dosáhnete toho pomocí StreamSequenceToken, neprůhledného IComparable objektu použitého k seřazení událostí. Producent může volání předat volitelnému StreamSequenceTokenOnNextAsync . Tento StreamSequenceToken proces se předává spotřebiteli a je dodáván s událostí. Tímto způsobem může vaše aplikace zpracovat a rekonstruovat své pořadí nezávisle na streamovacím modulu runtime.

Převinutelné streamy

Některé streamy umožňují přihlášení k odběru pouze od nejnovějšího bodu v čase, zatímco jiné umožňují "vrátit se v čase". Tato funkce závisí na základní technologii řízení front a na konkrétním poskytovateli datových proudů. Fronty Azure například umožňují využívat pouze nejnovější události ve frontě, zatímco Event Hubs umožňuje přehrání událostí z libovolného bodu v čase (až do určité doby vypršení platnosti). Datové proudy podporující návrat zpět v čase se nazývají převinutelné datové proudy.

Příjemce převinutelného datového proudu může předat StreamSequenceToken volání SubscribeAsync . Modul runtime doručuje události počínaje od StreamSequenceToken. Token null znamená, že příjemce chce přijímat události od nejnovější verze.

Schopnost převinout datový proud zpět je velmi užitečná ve scénářích obnovení. Představte si například entitu, která se přihlásí k odběru datového proudu a pravidelně ukládá svůj stav spolu s nejnovějším tokenem sekvence. Při zotavení ze selhání se může úloha znovu přihlásit ke stejnému streamu z nejnovějšího tokenu sekvence kontrolního bodu a obnovit bez ztráty událostí vygenerovaných od posledního kontrolního bodu.

Zprostředkovatel služby Event Hubs je přetáčitelný. Jeho kód najdete na GitHubu: Orleans/Azure/Orleans. Streaming.EventHubs. SMS (nyní vysílací kanál) a poskytovatelé front Azure nelze převinout zpět.

Bezstavové automaticky škálované zpracování horizontálního navýšení kapacity

Streamování ve výchozím nastavení Orleans podporuje velký počet relativně malých datových proudů, z nichž každý zpracovává jedno nebo více stavových zrn. Zpracování všech datových proudů je souhrnně rozděleno mezi mnoho standardních (stavových) částic. Kód aplikace řídí toto rozčlenění přiřazením ID datových proudů a ID zrna a explicitním přihlášením k odběru. Cílem je horizontálně dělené stavové zpracování.

Existuje ale také zajímavý scénář automatického bezstavového zpracování s horizontálním navýšením kapacity. V tomto scénáři má aplikace malý počet datových proudů (nebo dokonce jeden velký datový proud) a cílem je bezstavové zpracování. Příkladem je globální datový proud událostí, kdy zpracování zahrnuje dekódování každé události a potenciálně ji přesměrovává do jiných datových proudů pro další stavové zpracování. Bezstavové zpracování streamů s horizontálním navýšením kapacity je možné podporovat prostřednictvím OrleansStatelessWorkerAttribute zrn.

Aktuální stav bezstavového automatického škálování: To ještě není implementováno. Pokus o přihlášení k odběru datového StatelessWorkerAttribute proudu z agregace má za následek nedefinované chování. Zvažujeme podporu této možnosti.

Zrna a Orleans klienti

Orleans datové proudy fungují jednotně napříč zrny a Orleans klienty. To znamená, že stejná rozhraní API můžete použít uvnitř zrna a v Orleans klientovi k vytváření a využívání událostí. To výrazně zjednodušuje logiku aplikace, čímž se stávají speciální rozhraní API na straně klienta, jako jsou Grain Observers, zbytečnými.

Plně spravované a spolehlivé streamování pub-sub

Ke sledování odběrů streamů Orleans používá komponentu modulu runtime s názvem Streaming Pub-Sub, která slouží jako bod setkání pro uživatele a producenty streamů. Pub-sub sleduje všechna předplatná streamů, ukládá je a propojuje příjemce streamů s producenty streamů.

Aplikace můžou zvolit, kde a jak se ukládají data Pub-Sub. Samotná komponenta Pub-Sub je implementována jako zrnka (označovaná PubSubRendezvousGrain), která používá Orleans deklarativní trvalost. PubSubRendezvousGrain používá zprostředkovatele úložiště s názvem PubSubStore. Stejně jako u jakéhokoli agregačního intervalu můžete určit implementaci pro poskytovatele úložiště. Pro streaming Pub-Sub můžete pomocí konstruktoru hostitele sila změnit implementaci PubSubStore během sestavování sila:

Následující konfiguruje Pub-Sub k uložení jeho stavu do tabulek Azure.

var endpoint = new Uri(configuration["AZURE_TABLE_STORAGE_ENDPOINT"]!);
var credential = new DefaultAzureCredential();

hostBuilder.UseOrleans(siloBuilder =>
{
    siloBuilder.AddAzureTableGrainStorage("PubSubStore",
        options => options.TableServiceClient = new TableServiceClient(endpoint, credential));
});

hostBuilder.AddAzureTableGrainStorage("PubSubStore",
    options => options.ConnectionString = "<Secret>");

Tímto způsobem jsou Pub-Sub data trvale uložená v tabulce Azure. Pro počáteční vývoj můžete také použít úložiště paměti. Kromě pub-Sub Orleans poskytuje modul runtime streamování události od producentů uživatelů, spravuje všechny prostředky modulu runtime přidělené aktivně používaným datovým proudům a transparentně shromažďuje prostředky modulu runtime z nepoužívaných datových proudů.

Konfigurace

Pokud chcete používat streamy, musíte povolit poskytovatele datových proudů prostřednictvím hostitele sila nebo tvůrců klientů clusteru. Ukázkové nastavení zprostředkovatele streamu:

var tableEndpoint = new Uri(configuration["AZURE_TABLE_STORAGE_ENDPOINT"]!);
var queueEndpoint = new Uri(configuration["AZURE_QUEUE_STORAGE_ENDPOINT"]!);
var credential = new DefaultAzureCredential();

hostBuilder.UseOrleans(siloBuilder =>
{
    siloBuilder.AddMemoryStreams("StreamProvider")
        .AddAzureQueueStreams("AzureQueueProvider",
            optionsBuilder => optionsBuilder.ConfigureAzureQueue(
                options => options.Configure(
                    opt => opt.QueueServiceClient = new QueueServiceClient(queueEndpoint, credential))))
        .AddAzureTableGrainStorage("PubSubStore",
            options => options.TableServiceClient = new TableServiceClient(tableEndpoint, credential));
});

hostBuilder.AddSimpleMessageStreamProvider("SMSProvider")
    .AddAzureQueueStreams<AzureQueueDataAdapterV2>("AzureQueueProvider",
        optionsBuilder => optionsBuilder.Configure(
            options => options.ConnectionString = "<Secret>"))
    .AddAzureTableGrainStorage("PubSubStore",
        options => options.ConnectionString = "<Secret>");

Viz také

Orleans Zprostředkovatelé streamu

  • Poslední aktualizace v