Orleans detaljer om streams-implementering

Det här avsnittet innehåller en översikt över Orleans Stream-implementeringen på hög nivå. Den beskriver begrepp och information som inte är synliga på programnivå. Om du bara planerar att använda strömmar behöver du inte läsa det här avsnittet.

Terminologi:

Vi refererar med ordet "kö" till alla hållbara lagringsteknik som kan mata in strömhändelser och som antingen kan hämta händelser eller tillhandahåller en push-baserad mekanism för att använda händelser. För att ge skalbarhet tillhandahåller dessa tekniker vanligtvis shardade/partitionerade köer. Med Azure Queues kan du till exempel skapa flera köer och Event Hubs har flera hubbar.

Beständiga strömmar

Alla Orleans beständiga dataströmleverantörer delar en gemensam implementering PersistentStreamProvider. Dessa generiska stream-leverantörer måste konfigureras med en teknikspecifik IQueueAdapterFactory.

I testsyfte har vi till exempel köningsadapter som genererar sina testdata i stället för att läsa data från en kö. Koden nedan visar hur vi konfigurerar en beständig strömleverantör för att använda vår anpassade köadapter (generator). Det gör den genom att konfigurera den beständiga strömprovidern med en fabriksfunktion som används för att skapa adaptern.

hostBuilder.AddPersistentStreams(
    StreamProviderName, GeneratorAdapterFactory.Create);

När en strömproducent genererar ett nytt objekt och anropar stream.OnNext(), anropar strömningskörningen Orleans den lämpliga metoden på IQueueAdapter hos den strömleverantör som placerar objektet direkt i rätt kö.

Inhämta agenter

I hjärtat av den beständiga strömleverantören finns dragande agenter. Hämtagenter hämtar händelser från en uppsättning varaktiga köer och levererar dem till programkoden i enheter som förbrukar dem. Man kan se dragningsagentarna som en distribuerad "mikrotjänst" – en partitionerad, distribuerad, högtillgänglig och elastisk komponent. Dragande agenter körs inuti samma silor som innehåller applikationskorn och hanteras helt av Orleans Streaming Runtime.

StreamQueueMapper och StreamQueueBalancer

Hämtningsagenter parametriseras med IStreamQueueMapper och IStreamQueueBalancer. IStreamQueueMapper Innehåller en lista över alla köer och ansvarar även för att mappa strömmar till köer. På så sätt vet producentsidan för Persistent Stream-providern i vilken kö meddelandet ska skickas.

IStreamQueueBalancer uttrycker hur köer balanseras mellan silor och agenter. Målet är att tilldela köer till agenter på ett balanserat sätt för att förhindra flaskhalsar och stödja elasticitet. När en ny silo läggs till i Orleans klustret balanseras köerna automatiskt om över gamla och nya silor. Det StreamQueueBalancer tillåter anpassning av den processen. Orleans har flera inbyggda StreamQueueBalancers för att stödja olika balansscenarier (stort och litet antal köer) och olika miljöer (Azure, lokalt, statiskt).

Med testgeneratorexemplet ovan visar koden nedan hur man kan konfigurera kömapparen och köbalanseraren.

hostBuilder
    .AddPersistentStreams(StreamProviderName, GeneratorAdapterFactory.Create,
        providerConfigurator =>
        providerConfigurator.Configure<HashRingStreamQueueMapperOptions>(
            ob => ob.Configure(options => options.TotalQueueCount = 8))
      .UseDynamicClusterConfigDeploymentBalancer());

Koden ovan konfigurerar GeneratorAdapterFactory att använda en kömappare med åtta köer och balanserar köerna i klustret med hjälp av DynamicClusterConfigDeploymentBalancer.

Uttagningsprotokoll

Varje silo kör en uppsättning av hämtagenter, varje agent hämtar från en kö. Själva agenterna som hämtar implementeras av en intern körningskomponent, kallad SystemTarget. SystemTargets är i huvudsak körningskorn, omfattas av enkeltrådad samtidighet, kan använda vanliga kornmeddelanden och är lika lätta som korn. Till skillnad från grain är SystemTargets inte virtuella: de skapas uttryckligen (av körsystemet) och är inte platstransparenta. Genom att implementera dragningsagenter som SystemTargets Orleans kan Streaming Runtime förlita sig på inbyggda Orleans funktioner och kan skalas till ett mycket stort antal köer, eftersom det är lika billigt att skapa en ny dragningsagent som att skapa ett nytt korn.

Varje hämtningsagent kör en periodisk timer som hämtar från kön genom att anropa metoden IQueueAdapterReceiver.GetQueueMessagesAsync. De returnerade meddelandena placeras i den interna datastrukturen per agent med namnet IQueueCache. Varje meddelande inspekteras för att identifiera målströmmen. Agenten använder Pub-Sub för att ta reda på listan över strömkonsumenter som prenumererar på den här strömmen. När konsumentlistan har hämtats lagrar agenten den lokalt (i pub-sub-cachen) så att den inte behöver rådgöra med Pub-Sub i varje meddelande. Agenten prenumererar också på pub-sub för att få meddelande om nya konsumenter som prenumererar på den strömmen. Det här handslaget mellan agenten och pub-sub garanterar stark strömningsprenumerationssemantik: när konsumenten har prenumererat på strömmen, kommer den att se alla händelser som genererades efter att den har prenumererat. Dessutom tillåter StreamSequenceToken prenumeration vid en tidigare tidpunkt.

Köcache

IQueueCache är en intern datastruktur per agent som möjliggör att skilja borttagandet av nya händelser från kön och leverera dem till konsumenter. Det möjliggör också avskiljning av leverans till olika strömmar och olika användare.

Föreställ dig en situation där en ström har 3 strömkonsumenter och en av dem är långsam. Om försiktighet inte vidtas kan den här långsamma konsumenten påverka agentens framsteg, vilket saktar ned konsumtionen för andra konsumenter av den strömmen och till och med saktar ned köhanteringen och leveransen av händelser för andra strömmar. För att förhindra detta och tillåta maximal parallellitet i agenten använder IQueueCachevi .

IQueueCache buffrar streamhändelser och ger agenten ett sätt att leverera händelser till varje konsument i sin egen takt. Leveransen per konsument implementeras av den interna komponenten som heter IQueueCacheCursor, som spårar förloppet per konsument. På så sätt tar varje konsument emot händelser i sin egen takt: snabba konsumenter får händelser så snabbt som de blir av med kön, medan långsamma konsumenter får dem senare. När meddelandet har levererats till alla konsumenter kan det tas bort från cacheminnet.

Ryggtryck

Backpressure i Orleans Streaming Runtime gäller på två platser: att överföra strömhändelser från kön till agenten och leverera händelserna från agenten till strömkonsumenter.

Det senare tillhandahålls av den inbyggda Orleans mekanismen för meddelandeleverans. Varje strömhändelse levereras från agenten till konsumenter via standardmeddelandet Orleans för korn, en i taget. Det vill säga att agenterna skickar en händelse (eller en begränsad mängd händelser) till varje strömkonsument och väntar på det här anropet. Nästa händelse börjar inte levereras förrän uppgiften för den föregående händelsen har lösts eller avslutats. På så sätt begränsar vi naturligtvis leveransfrekvensen per konsument till ett meddelande i taget.

När du tar med dataströmhändelser från kön till agenten Orleans tillhandahåller Streaming en ny speciell mekanism för mottryck. Eftersom agenten frikopplar avkonteringen av händelser från kön och levererar dem till konsumenter, kan en enda långsam konsument hamna efter så mycket att IQueueCache fylls upp. För att förhindra att IQueueCache den växer på obestämd tid begränsar vi dess storlek (storleksgränsen kan konfigureras). Agenten kastar dock aldrig bort händelser som inte har levererats.

När cacheminnet i stället börjar fyllas på saktar agenterna ned frekvensen för att dequeuera händelser från kön. På så sätt kan vi "rida ut" de långsamma leveransperioderna genom att justera den hastighet med vilken vi konsumerar från kön ("backpressure") och komma tillbaka till snabba förbrukningsfrekvenser senare. För att identifiera "långsam leverans"-dalar använder IQueueCache en intern datastruktur av cache-buckets som spårar förloppet för leverans av händelser till individuella strömkonsumenter. Detta resulterar i ett mycket dynamiskt och självjusterande system.