Prímszámok adatfolyamként, RxJS-sel és Go-val

Az RxJs-ben a magasabb szintű megfigyelhető elemek egyenlítését olyan operátorok hajtják végre, mint a mergeMap , switchMap , exaustMap és concatMap . Ebben az esetben a concatMap értéket használjuk, mivel az értesítésnek azonnal el kell indulnia, amint az előző Megfigyelhető forrás (a fenti példában obsFirst) befejeződik.

Ha a pNumbers rekurzív függvény készen áll, lehetséges a prímszámok kiszámítása egy bizonyos küszöbértékig:

Létrehozhatunk egy Observable-et is, amely minden prímszámot értesít egy adott küszöbértékig. Ezt egy új Observable burkolást pNumbers létrehozó függvényen keresztül kapjuk meg.

A fent meghatározott primeNumbers függvény által visszaadott Observable teszteléséhez csak elő kell fizetnünk rá, így:

const pNums = primeNumbers(100);
pNums.subscribe(pn => console.log(pn));

Ez csak a szűrők hosszú láncolata, amely végül eléri a „maximális hívási verem méretét”

Ha alaposan megnézzük a megvalósításunkat, rájövünk, hogy a logika lényege a következő sorokban van:

return pNumbers(source.pipe(
  filter((n) => {
    return n % primeNumber !== 0;
  })
);

Itt láncoljuk a source adatfolyamot egy filter függvénnyel. Mivel ezt a sort rekurzívan hívják, végül egy hosszú filter függvényláncot építünk fel, amely a számításunk kezdeti source adatfolyamához kapcsolódik.

Ha elégszer megismételjük ezt a rekurziót, akkor sok filter függvényt egymás után láncolunk, és végül elérjük a hívási verem maximális méretét. Pontosan ez történik, ha ezt a kódot elég nagy upTo bemeneti számmal futtatjuk.

„Eratosthenes szitája” Go csatornák használata folyamként

A Go-ban csatornákat használhatunk streamek megvalósítására. Például 2-től kezdődő kötetlen egész számok folyama generálható így:

ch := make(chan int)
for i := 2; ; i++ {
  ch <- i
}

Ebben az esetben a streamet a ch csatorna képviseli. Bármely ügyfélkódnak, amely olvasni akarja az adatfolyamot, csak a ch csatornát kell olvasnia.

Logikánk arra kéri, hogy rekurzív módon hozzunk létre egy prímszám nem többszöröseiből álló adatfolyamokat, amelyek egy bemeneti adatfolyamból indulnak ki. Kódolhatunk egy filter függvényt az ilyen logika megvalósításához a csatorna-ként koncepció használatával:

func filter(inStream <-chan int, filteredStream chan<- int, n int) {
  for i := range inStream {
    if i%n != 0 {
      filteredStream <- i
    }
  }
}

A inStream a bemeneti adatfolyamot reprezentáló csatorna, az filteredStream pedig az n nem többszöröseinek folyamát reprezentáló csatorna, ahol n a prímszám. Vegye figyelembe az <-chan és chan<- használatát. Az inStream típusa <-chan int, ami egy olyan csatornát jelent, amelyről csak olvasni tudunk, míg az fitleredStream chan<- int típusú, ami azt jelenti, hogy ez egy olyan csatorna, ahol csak írni tudunk.

Végül szükségünk van az pNumbers függvényre, amely egy bemeneti adatfolyam esetén felveszi az első értéket, ami egy prímszám, alkalmazza a szűrési logikát a folyam többi részére, hogy új „szűrt” adatfolyamot hozzon létre, és rekurzívan hívja meg magát ezzel az új „ szűrt” adatfolyam bemeneti adatfolyamként:

func pNumbers(inStream <-chan int, lastPrime int) {
  primeNumber := <-inStream 
  fmt.Println(i, primeNumber)
  filteredStream := make(chan int)
  go filter(inStream, filteredStream, primeNumber)
  pNumbers(filteredStream, primeNumber)
}

A teljes megoldás végleges változata csak az általunk most elkészített építőelemek kompozíciója.

Érdemes megjegyezni, hogy a pNumbers függvény minden rekurzív hívása ugyanabban a „fő” gorutinban fut, míg a generate és filter hívások mindegyike a saját gorutinjában fut. Ennek alapvető oka az, hogy a pNumbers csak fogyasztója a folyamoknak, más szóval csak olvas a létrehozott csatornákból, és szekvenciálisan kell működnie, mivel prímszámok rendezett sorozatát akarjuk. A fogyasztónak termelőkre van szüksége, vagyis olyan funkciókra, amelyek a csatornákon írnak. A generate producer, mivel csak egyetlen csatornán ír. A filter fogyasztó és termelő, hiszen az egyik csatornán olvas, a másikon ír. Ugyanazon adatfolyam fogyasztóinak és előállítóinak, azaz az ugyanazon a csatornán lévő olvasóknak és íróknak külön gorutinokban kell futniuk, hogy elkerüljék a holtpontokat.

Az 1 utáni első három prímszám (azaz 2, 3, 5) gorutinjai és csatornái közötti különféle interakciókat az alábbiakban szemléltetjük.

A tényleges olvasási és írási sorrend kissé eltérhet. Például a gr2 gorutin 3-at írhat a forrás chan-ba, mielőtt a gr1 létrehozná a 2-es csatornát. Logikai szempontból azonban a fenti diagram világosan ábrázolja a folyamat során lejátszódó interakciókat.

Érdekes megnézni a számítás lezárásának módját is. A leállítás úgy kezdődik, hogy a generate függvény kilép a ciklusából, amikor elérte a upTo értéket, majd bezárja a csatornaforrás chanját, azaz azt a csatornát, amelyre ír. A source chan az a csatorna, amelyről a gr3 goourine-ban futó filter függvény olvas. Tehát, amikor ez a függvény megpróbálja a következő olvasást a source chan-től, zárt csatornát talál, ezért kilép a for rangehurokból, és bezárja a chan 2 csatornát, azaz azt a csatornát, ahol ír. nak nek. A chan 2 az a csatorna, amelyről a gr4-ben futó filter függvény olvas. De most chan 2lezárult, ami a 3. csatorna bezárásához vezet. És így tovább és így tovább, amíg el nem érjük a pNumbers által létrehozott utolsó csatornát, azaz a gr1 csatorna olvasni akar. De ez az olvasási művelet nem fejeződhet be, mivel a csatorna bezárul, mielőtt bármit is írnának bele, így a pNumbers kilép, aminek következtében az egész program kilép.

Ez a sorrend itt látható:

A fenti megvalósítás a Go "Prime Sieve" stream koncepciójára összpontosító változata, amely ugyanannak a problémának egy nagyon elegáns megoldása, amelyet ebben az "érdekes előadásban a Go miértjeiről" találhat meg.

A párhuzamosság szépsége abban rejlik, hogy kihasználhatja az összes magját

Az „Eratosthenes szita” Go megvalósítása nagymértékben kihasználja az egyidejűséget. Az első X prímszám megtalálásához (kettőtől kezdve) X gorutint hozunk létre, amelyek párhuzamosan futnak. A logikát párhuzamosan szerveztük, hogy egyszerre sok mindennel tudjunk foglalkozni. Ez a sok dolog a különböző szűrési műveletek, amelyek párhuzamosan zajlanak.

A végrehajtás nem feltétlenül párhuzamos, abban az értelemben, hogy csak egy elérhető maggal is működik. De ha sok mag van, akkor a párhuzamosság nagyon természetesen levezethető. A Go futtatókörnyezet gondoskodik arról, hogy az összes magot felhasználják, csökkentve a végrehajtási időt, és a magok száma nő.

Következtetések

A stream alapú logika két különböző megvalósítását láttuk az „Eratosthenes szita” algoritmus alapján, az egyik RxJs Observable és egy Go csatornákat használ.

Nem az volt a cél, hogy hatékony implementációkat építsünk prímszámok keresésére, hanem az volt, hogy átérezzük a stream alapú logika erejét és azt a természetes megvalósítást, amelyet az RxJ és a Go csatornák is biztosítanak ehhez a programozási modellhez.

Az adatfolyam-alapú logika valószínűleg egyre elterjedtebb lesz a „felhő, elosztott számítástechnika és eseményvezérelt architektúrák” által uralt világban. Jobb, ha felkészülünk, és megtanuljuk használni ezeket az eszközöket.