Orleans rychlý start streamování

Tato příručka vám ukáže rychlý způsob, jak nastavit a používat Orleans Streamy. Další informace o funkcích streamování najdete v dalších částech této dokumentace.

Požadované konfigurace

V této příručce použijete datový proud založený na paměti, který používá zasílání odstupňovaných zpráv k odesílání streamových dat odběratelům. K ukládání seznamů předplatných použijete poskytovatele úložiště v paměti. Použití mechanismů založených na paměti pro streamování a úložiště je určeno pouze pro místní vývoj a testování, nikoli pro produkční prostředí.

Orleans streamování vyžaduje balíček Microsoft.Orleans.Streaming NuGet Tento balíček poskytuje funkce streamování pro klienta i server, včetně AddMemoryStreams metody rozšíření použité v této příručce.

Na silu, kde silo je ISiloBuilder, zavolejte AddMemoryStreams.

silo.AddMemoryStreams("StreamProvider")
    .AddMemoryGrainStorage("PubSubStore");

Na klientovi clusteru, kde client je IClientBuilder, zavolejte AddMemoryStreams.

client.AddMemoryStreams("StreamProvider");

V této příručce použijte jednoduchý datový proud založený na zprávách, který používá odstupňované zasílání zpráv k odesílání streamových dat odběratelům. Použití poskytovatele úložiště v paměti k ukládání seznamů předplatných; to není moudrá volba pro skutečné produkční aplikace.

Na silu, kde hostBuilder je ISiloHostBuilder, zavolejte AddSimpleMessageStreamProvider.

hostBuilder.AddSimpleMessageStreamProvider("SMSProvider")
           .AddMemoryGrainStorage("PubSubStore");

Na klientovi clusteru, kde clientBuilder je IClientBuilder, zavolejte AddSimpleMessageStreamProvider.

clientBuilder.AddSimpleMessageStreamProvider("SMSProvider");

Poznámka:

Ve výchozím nastavení jsou zprávy předávané přes Simple Message Stream považovány za neměnné a mohou být předány odkazem na jiná zrnka. Chcete-li toto chování vypnout, nakonfigurujte poskytovatele serveru SMS tak, aby vypnul SimpleMessageStreamProviderOptions.OptimizeForImmutableData.

siloBuilder
    .AddSimpleMessageStreamProvider(
        "SMSProvider",
        options => options.OptimizeForImmutableData = false);

Můžete vytvářet datové proudy, odesílat je jako producenti a přijímat data jako odběratelé.

Produkovat události

Vytváření událostí pro datové proudy je poměrně snadné. Nejprve získejte přístup k poskytovateli streamu definovanému v konfiguraci dříve ("StreamProvider"), pak zvolte datový proud a nasdílejte do něj data.

// Pick a GUID for a chat room grain and chat room stream
var guid = new Guid("some guid identifying the chat room");
// Get one of the providers which we defined in our config
var streamProvider = GetStreamProvider("StreamProvider");
// Get the reference to a stream
var streamId = StreamId.Create("RANDOMDATA", guid);
var stream = streamProvider.GetStream<int>(streamId);

Vytváření událostí pro datové proudy je poměrně snadné. Nejprve získejte přístup k poskytovateli streamu definovanému v konfiguraci dříve ("SMSProvider"), pak zvolte datový proud a nasdílejte do něj data.

// Pick a GUID for a chat room grain and chat room stream
var guid = new Guid("some guid identifying the chat room");
// Get one of the providers which we defined in our config
var streamProvider = GetStreamProvider("SMSProvider");
// Get the reference to a stream
var stream = streamProvider.GetStream<int>(guid, "RANDOMDATA");

Jak vidíte, stream má identifikátor GUID a obor názvů. To usnadňuje identifikaci jedinečných datových proudů. Jmenný prostor pro chatovací místnost může být například "Místnosti", a identifikátor GUID může být identifikátor GUID vlastníka RoomGrain.

Tady použijte identifikátor GUID známé chatovací místnosti. Pomocí metody OnNextAsync datového proudu do něj odešlete data. Pojďme to udělat uvnitř časovače pomocí náhodných čísel. Pro datový proud můžete použít i jakýkoli jiný datový typ.

RegisterTimer(_ =>
{
    return stream.OnNextAsync(Random.Shared.Next());
},
null,
TimeSpan.FromMilliseconds(1_000),
TimeSpan.FromMilliseconds(1_000));

Přihlášení k odběru a přijímání streamovaných dat

Pro příjem dat můžete použít implicitní a explicitní předplatná popsaná podrobněji u explicitních a implicitních předplatných. V tomto příkladu se používají implicitní předplatná, která jsou jednodušší. Pokud se typ zrnitosti chce implicitně přihlásit k odběru datového proudu, používá atribut [ImplicitStreamSubscription(namespace)].

Pro váš případ definujte ReceiverGrain takto:

[ImplicitStreamSubscription("RANDOMDATA")]
public class ReceiverGrain : Grain, IRandomReceiver

Vždy, když se data odsílají do proudů v prostoru názvů RANDOMDATA (jako v příkladu časovače), obdrží zprávu objekt typu ReceiverGrain se stejným Guid jako proud. I když v současné době neexistují žádné aktivace zrna, modul runtime automaticky vytvoří nové a odešle do něj zprávu.

Aby to fungovalo, dokončete proces předplatného nastavením OnNextAsync metody pro příjem dat. Chcete-li tak učinit, ReceiverGrain by měl volat něco podobného ve svém OnActivateAsync:

// Create a GUID based on our GUID as a grain
var guid = this.GetPrimaryKey();

// Get one of the providers which we defined in config
var streamProvider = GetStreamProvider("StreamProvider");

// Get the reference to a stream
var streamId = StreamId.Create("RANDOMDATA", guid);
var stream = streamProvider.GetStream<int>(streamId);

// Set our OnNext method to the lambda which simply prints the data.
// This doesn't make new subscriptions, because we are using implicit
// subscriptions via [ImplicitStreamSubscription].
await stream.SubscribeAsync<int>(
    async (data, token) =>
    {
        Console.WriteLine(data);
        await Task.CompletedTask;
    });

// Create a GUID based on our GUID as a grain
var guid = this.GetPrimaryKey();

// Get one of the providers which we defined in config
var streamProvider = GetStreamProvider("SMSProvider");

// Get the reference to a stream
var stream = streamProvider.GetStream<int>(guid, "RANDOMDATA");

// Set our OnNext method to the lambda which simply prints the data.
// This doesn't make new subscriptions, because we are using implicit
// subscriptions via [ImplicitStreamSubscription].
await stream.SubscribeAsync<int>(
    async (data, token) =>
    {
        Console.WriteLine(data);
        await Task.CompletedTask;
    });

Máte všechno připravené! Nyní jediným požadavkem je, že něco aktivuje vytvoření producentního zrna. Pak zaregistruje časovač a začne odesílat náhodná celá čísla všem zúčastněným stranám.

Tento průvodce znovu přeskočí mnoho podrobností a poskytuje pouze základní přehled. Přečtěte si další části této příručky a dalších zdrojů informací o Rx, abyste získali dobrou představu o tom, co je k dispozici a jak funguje.

Reaktivní programování může být účinným přístupem k řešení mnoha problémů. LinQ můžete například v odběrateli použít k filtrování čísel a provádění různých zajímavých operací.

Viz také

Orleans Programovací rozhraní API pro toky