Comment puis-je imprimer un flux continu de messages de Twitter avec quelques lignes de code en ajoutant des données météorologiques aux endroits où vivent leurs auteurs? Et comment pouvez-vous limiter la vitesse des demandes au fournisseur météo pour qu'il ne nous liste pas?

Aujourd'hui, nous vous expliquerons comment procéder, mais nous découvrirons tout d'abord la technologie Akka Streams, qui rend le travail avec les flux de données en temps réel aussi simple que les programmeurs travaillant avec des expressions LINQ sans nécessiter la mise en œuvre d'acteurs individuels ou d'interfaces Reactive Streams .


L'article est basé sur une transcription du rapport de Vagif Abilov de notre conférence de décembre DotNext 2017 Moscou.

Je m'appelle Vagif, je travaille pour la société norvégienne Miles. Aujourd'hui, nous allons parler de la bibliothèque Akka Streams.

Akka et Reactive Streams sont l'intersection d'ensembles assez étroits, et on pourrait avoir l'impression que c'est une niche telle que vous devez avoir une grande connaissance pour entrer, mais juste le contraire. Et cet article est destiné à montrer qu'en utilisant Akka Streams, vous pouvez éviter la programmation de bas niveau qui est requise lors de l'utilisation de Reactive Streams et Akka.NET. Pour l'avenir, je peux dire immédiatement: si au tout début de notre projet, sur lequel nous utilisons Akka, nous connaissions l'existence d'Akka Streams, nous écririons beaucoup différemment, nous économiserions à la fois du temps et du code.

"Peut-être que le pire que vous puissiez faire est d'amener les personnes qui ne souffrent pas à prendre votre aspirine."
Max Kreminski
«Portes fermées, maux de tête et besoins intellectuels»

Avant d'entrer dans les détails techniques, un peu sur ce que votre chemin vers Akka Streams peut s'avérer être et ce qui peut vous y conduire. Un jour, je suis tombé sur le blog de Max Kreminski, où il a posé une telle question philosophique aux programmeurs: comment ou pourquoi il est impossible pour un programmeur d'expliquer ce que sont les monades. Il l'a expliqué de cette façon: très souvent, les gens vont immédiatement aux détails techniques, expliquant à quel point la programmation est magnifiquement fonctionnelle et à quel point il y a du sens dans la monade, sans prendre la peine de se demander pourquoi le programmeur pourrait en avoir besoin. Dessiner une analogie, c'est comme essayer de vendre de l'aspirine sans se soucier de savoir si votre patient souffre.

En utilisant cette analogie, je voudrais poser la question suivante: si Akka Streams est de l'aspirine, quelle devrait être la douleur qui vous y conduira?

Flux de données

Parlons d'abord des flux de données. Le flux peut être assez simple, linéaire.


Ici, nous avons un certain consommateur de données (un lapin dans la vidéo). Il consomme des données à une vitesse qui lui convient. Il s'agit de l'interaction idéale du consommateur avec le flux: elle établit la bande passante et les données y circulent tranquillement. Ce flux de données simple peut être infini ou se terminer.

Mais le flux peut être plus complexe. Si vous plantez plusieurs lapins côte à côte, nous aurons déjà une parallélisation des flux. Ce que Reactive Streams essaie de résoudre, c'est précisément comment nous pouvons communiquer avec les flux à un niveau plus conceptuel, c'est-à-dire, que nous parlions simplement d'une sorte de mesure de capteur de température, où les mesures linéaires entrent en jeu. , ou nous avons des mesures en continu de milliers de capteurs de température entrant dans le système via des files d'attente RabbitMQ et stockés dans les journaux système. Tout ce qui précède peut être considéré comme un seul flux composite. Si vous allez encore plus loin, la gestion automatisée de la production (par exemple, par une boutique en ligne) peut également être réduite à un flux de données, et ce serait formidable si nous pouvions parler de la planification d'un tel flux, quelle que soit sa complexité.



Pour les projets modernes, le support des threads n'est pas très bon. Si je me souviens bien, Aaron Stannard, dont vous voyez le tweet dans l'image, voulait obtenir un flux d'un fichier de plusieurs gigaoctets contenant CSV, c'est-à-dire texte, et il s'est avéré qu'il n'y a rien que vous pouvez simplement prendre et utiliser immédiatement, sans un tas d'actions supplémentaires. Mais il n'a tout simplement pas pu obtenir un flux de valeurs CSV, ce qui l'a attristé. Il y a peu de solutions (à l'exception de certaines zones spéciales), beaucoup est réalisé par les anciennes méthodes, lorsque nous ouvrons tout cela, commençons la lecture, la mise en mémoire tampon, dans le pire des cas, nous obtenons quelque chose comme le bloc-notes, qui dit que le fichier est trop gros.

À un niveau conceptuel élevé, nous sommes tous engagés dans le traitement des flux de données, et Akka Streams vous aidera si:

• Vous connaissez Akka, mais vous voulez vous épargner les détails associés à l'écriture du code d'acteur et à sa coordination;
  
• Vous connaissez les flux réactifs et souhaitez utiliser une implémentation prête à l'emploi de leurs spécifications;
  
• Les éléments de bloc Akka Streams pour les étapes conviennent à la modélisation de votre processus;
  
• Vous souhaitez profiter de la contre-pression (contre-pression) d'Akka Streams pour gérer et affiner dynamiquement les étapes de débit de votre flux de travail.
  

Des acteurs aux Akka Streams

La première façon est celle des acteurs aux Akka Streams, à ma façon.

La photo montre pourquoi nous avons commencé à utiliser le modèle d'acteur. Nous étions épuisés par le contrôle manuel des flux, l'état partagé, c'est tout. Tous ceux qui ont travaillé avec de grands systèmes, avec plusieurs threads, comprennent combien cela prend du temps et combien il est facile de s'y tromper, ce qui peut être fatal pour l'ensemble du processus. Cela nous a conduit au modèle des acteurs. Nous ne regrettons pas le choix fait, mais, bien sûr, lorsque vous commencez à travailler et à programmer davantage, ce n'est pas que l'enthousiasme initial cède la place à autre chose, mais vous commencez à réaliser que quelque chose pourrait être fait encore plus efficacement.

«Par défaut, les destinataires de leurs messages sont entrés dans le code des acteurs. Si je crée un acteur A qui envoie un message à l'acteur B et que vous souhaitez remplacer le destinataire par l'acteur C, dans le cas général, cela ne fonctionnera pas pour vous. »
Noel Welch (underscore.io)

Les acteurs critiqués pour ne pas avoir composé. L'un des premiers à écrire à ce sujet sur son blog a été Noel Welch, l'un des développeurs d'Underscore. Il a remarqué que le système d'acteurs ressemble à ceci:



Si vous n'utilisez aucun élément supplémentaire, comme une injection de dépendance, l'adresse de son destinataire est cousue dans l'acteur.



Quand ils commencent à s'envoyer des messages, vous définissez tout cela à l'avance, en programmant les acteurs. Et sans astuces supplémentaires, un tel système rigide est obtenu.
Un des développeurs d'Akka, Roland Kuhn, a expliqué ce que l'on entend généralement par mauvaise disposition. La méthode acteur est basée sur la méthode tell, c'est-à-dire les messages unidirectionnels: elle est de type void, c'est-à-dire qu'elle ne renvoie rien (ou unité, selon la langue). Il est donc impossible de construire une description du processus à partir d'une chaîne d'acteurs. Alors tu as envoyé dire, alors quoi? Arrêter Nous sommes devenus nuls. Vous pouvez le comparer, par exemple, avec des expressions LINQ, où chaque élément de l'expression renvoie IQueryable, IEnumerable, et tout cela peut être facilement compilé. Les acteurs ne donnent pas une telle opportunité. Dans le même temps, Roland Kuhn s'est opposé au fait qu'ils, disent-ils, ne composent pas en principe, affirmant qu'en fait ils sont compilés d'autres manières, dans le même sens où la société humaine se prête à l'agencement. Cela ressemble à un argument philosophique, mais si vous y réfléchissez, l'analogie est logique - oui, les acteurs s'envoient des messages unidirectionnels, mais nous communiquons également les uns avec les autres, en prononçant des messages unidirectionnels, mais en même temps, nous interagissons assez efficacement, c'est-à-dire que nous créons des systèmes complexes. Néanmoins, une telle critique des acteurs existe.

public class SampleActor : ReceiveActor { public SampleActor() { Idle(); } protected override void PreStart() { /* ... */ } private void Idle() { Receive<Job>(job => /* ... */); } private void Working() { Receive<Cancel>(job => /* ... */); } } 

De plus, l'implémentation de l'acteur nécessite au moins d'écrire une classe si vous travaillez en C #, ou des fonctions si vous travaillez en F #. Dans l'exemple ci-dessus - code passe-partout, que vous devez de toute façon écrire. Bien qu'il ne soit pas très volumineux, c'est un certain nombre de lignes que vous devrez toujours écrire à ce bas niveau. Presque tout le code qui est présent ici est une sorte de cérémonie. Ce qui se passe lorsqu'un acteur reçoit directement un message ne s'affiche pas du tout ici. Et tout cela doit être écrit. Bien sûr, ce n'est pas beaucoup, mais c'est la preuve que nous travaillons avec des acteurs de bas niveau, créant de telles méthodes de vide.

Et si nous pouvions passer à un niveau différent et supérieur, nous poser des questions sur la modélisation de notre processus, qui comprend le traitement de données provenant de diverses sources qui sont mélangées, converties et transférées?

 var results = db.Companies .Join(db.People, c => c.CompanyID, p => p.PersonID, (c, p) => new { c, p }) .Where(z => zcCreated >= fromDate) .OrderByDescending(z => zcCreated) .Select(z => zp) .ToList(); 

Un analogue de cette approche peut être ce que nous avons tous l'habitude de travailler avec LINQ depuis dix ans. Nous ne nous demandons pas comment fonctionne Join. Nous savons qu'il existe un tel fournisseur LINQ qui fera tout cela pour nous, et nous sommes intéressés à un niveau supérieur pour répondre à la demande. Et nous pouvons généralement mélanger les bases de données ici, nous pouvons envoyer des demandes distributives. Et si vous pouviez décrire le processus de cette façon?

 HttpGet pageUrl |> fun s -> Regex.Replace(s, "[^A-Za-z']", " ") |> fun s -> Regex.Split(s, " +") |> Set.ofArray |> Set.filter (fun word -> not (Spellcheck word)) |> Set.iter (fun word -> printfn " %s" word) 

(Source)

Ou, par exemple, des transformations fonctionnelles. Ce que beaucoup de gens aiment à propos de la programmation fonctionnelle, c'est que vous pouvez transmettre des données à travers une série de transformations, et vous obtenez un code compact assez clair, quelle que soit la langue dans laquelle vous l'écrivez. C'est assez facile à lire. Le code de l'image est spécialement écrit en F #, mais en général, probablement, tout le monde comprend ce qui se passe ici.

 val in = Source(1 to 10) val out = Sink.ignore val bcast = builder.add(Broadcast[Int](2)) val merge = builder.add(Merge[Int](2)) val f1,f2,f3,f4 = Flow[Int].map(_ + 10) source ~> f1 ~> bcast ~> f2 ~> merge ~> f3 ~> sink bcast ~> f4 ~> merge ~> 

(Source)

Et alors? Dans l'exemple ci-dessus, nous avons une source de données source, qui se compose d'entiers de 1 à 10. C'est ce qu'on appelle le DSL graphique (langage spécifique au domaine). Les éléments de la langue du domaine dans l'exemple ci-dessus sont des flèches unidirectionnelles - ce sont des opérateurs supplémentaires définis par des outils de langue qui montrent graphiquement la direction du flux. Nous passons Source à travers une série de transformations - pour faciliter la démonstration, elles ajoutent toutes un dix au nombre. Vient ensuite la diffusion: nous multiplions les canaux, c'est-à-dire que chaque numéro entre dans deux canaux. Ensuite, nous ajoutons à nouveau 10, mélangons nos flux de données, obtenons un nouveau flux, en ajoutons également 10, et tout cela va à notre flux de données, dans lequel rien ne se passe. Il s'agit du vrai code écrit en Scala, qui fait partie d'Akka Streams, implémenté dans ce langage. C'est-à-dire que vous spécifiez les phases de la transformation de vos données, indiquez quoi en faire, spécifiez la source, le stock, certains points de contrôle, puis formez un tel graphique à l'aide du DSL graphique. C'est tout le code pour un seul programme. Quelques lignes de code montrent ce qui se passe dans le processus.

Oublions comment écrire le code de définition pour les acteurs individuels et apprenons plutôt les primitives de mise en page de haut niveau qui créeront et connecteront les acteurs requis en eux-mêmes. Lorsque nous exécutons un tel graphique, le système qui fournit Akka Streams créera lui-même l'acteur requis, y enverra toutes ces données, les traitera comme il se doit et les transmettra finalement au destinataire final.

 var runnable = Source .From(Enumerable.Range(1, 1000)) .Via(Flow.Create<int>().Select(x => x * 2) .To(Sink.ForEach<int>(x => Console.Write(x.ToString)); 

L'exemple ci-dessus montre à quoi cela pourrait ressembler en C #. La manière la plus simple: nous avons une source de données - ce sont des nombres de 1 à 1000 (comme vous pouvez le voir, dans Akka Streams, tout IEnumerable peut devenir une source de flux de données, ce qui est très pratique). Nous faisons un calcul simple, disons, nous multiplions par deux, puis sur le flux de données tout cela est affiché à l'écran.

 var graph = GraphDsl.Create(builder => { var bcast = builder.Add(new Broadcast<int>(2)); var merge = builder.Add(new Merge<int, int>(2)); var count = Flow.FromFunction(new Func<int, int>(x => 1)); var sum = Flow.Create<int>().Sum((x, y) => x + y); builder.From(bcast.Out(0)).To(merge.In(0)); builder.From(bcast.Out(1)).Via(count).Via(sum).To(merge.In(1)); return new FlowShape<int, int>(bcast.In, merge.Out); }); 

Ce qui est illustré dans l'exemple ci-dessus est appelé «DSL graphique en C #». En fait, il n'y a pas de graphiques ici, c'est un port avec Scala, mais en C # il n'y a aucun moyen de définir les opérateurs de cette façon, donc cela semble un peu plus encombrant, mais toujours assez compact pour comprendre ce qui se passe ici. Donc, nous créons un certain graphique (il existe différents types de graphiques, ici on l'appelle FlowShape) à partir de différents composants, où il y a une source de données et il y a des transformations. Nous envoyons des données à un canal dans lequel nous générons le comptage, c'est-à-dire le nombre d'éléments de données à transmettre, et dans l'autre, nous générons la somme, puis nous mélangons le tout. Ensuite, nous verrons des exemples plus intéressants que le simple traitement de nombres entiers.
C'est le premier chemin qui peut vous conduire à Akka Streams, si vous avez de l'expérience avec un modèle d'acteur et que vous avez pensé à écrire manuellement chacun, même l'acteur le plus simple. La deuxième façon dont Akka Streams arrive à travers Reactive Streams.

Des flux réactifs aux flux Akka

Qu'est-ce que les flux réactifs ? Il s'agit d'une initiative conjointe visant à développer une norme pour le traitement asynchrone des flux de données. Il définit l'ensemble minimal d'interfaces, de méthodes et de protocoles qui décrivent les opérations et entités nécessaires pour atteindre l'objectif - traitement asynchrone des données en temps réel avec une contre-pression non bloquante (contre-pression). Il permet diverses implémentations utilisant différents langages de programmation.

Reactive Streams vous permet de traiter un nombre potentiellement illimité d'éléments dans une séquence et de transférer des éléments de manière asynchrone entre les composants avec une contre-pression non bloquante.

La liste des initiateurs de la création de Reactive Streams est assez impressionnante: voici Netflix, Oracle et Twitter.

La spécification est très simple pour rendre l'implémentation dans différentes langues et plates-formes aussi accessible que possible. Les principaux composants de l'API Reactive Streams:

1. Éditeur
   
2. Abonné
   
3. Abonnement
   
4. Processeur
   

Essentiellement, cette spécification n'implique pas que vous commencerez manuellement à implémenter ces interfaces. Il est entendu que certains développeurs de bibliothèques le feront pour vous. Et Akka Streams est l'une des implémentations de cette spécification.

 public interface IPublisher<out T> { void Subscribe(ISubscriber<T> subscriber); } public interface ISubscriber<in T> { void OnSubscribe(ISubscription subscription); void OnNext(T element); void OnError(Exception cause); void OnComplete(); } 

Les interfaces, comme vous pouvez le voir dans l'exemple, sont vraiment très simples: par exemple, Publisher ne contient qu'une seule méthode - «s'abonner». L'abonné, l'abonné, ne contient que quelques réactions à l'événement.

 public interface ISubscription { void Request(long n); void Cancel(); } public interface IProcessor<in T1, out T2> : ISubscriber<T1>, IPublisher<T2> { } 

Enfin, l'abonnement contient deux méthodes - «démarrer» et «refuser». Le processeur ne définit aucune nouvelle méthode, il associe un éditeur et un abonné.

Qu'est-ce qui distingue les flux réactifs des autres implémentations de flux? Reactive Streams combine des modèles push et pull. Pour le support, il s'agit du scénario de performances le plus efficace. Supposons que vous ayez un abonné lent aux données. Dans ce cas, pousser pour lui peut être fatal: si vous lui envoyez une énorme quantité de données, il ne pourra pas les traiter. Il est préférable d'utiliser pull pour que l'abonné récupère lui-même les données de l'éditeur. Mais si l'éditeur est lent, il s'avère que l'abonné est bloqué tout le temps, attendant tout le temps. Une solution intermédiaire peut être la configuration: nous avons un fichier de configuration dans lequel nous déterminons lequel est le plus rapide. Et si leurs vitesses changent?

Ainsi, l'implémentation la plus élégante est celle dans laquelle nous pouvons changer dynamiquement les modèles push et pull.


(Source (Apache Flink))

Le diagramme montre comment cela peut se produire. Cette démo utilise Apache Flink. Yellow est un éditeur, producteur de données, il était fixé à environ 50% de ses capacités. L'abonné essaie de choisir la meilleure stratégie - elle se révèle être push. Ensuite, nous avons réinitialisé l'abonné à une vitesse d'environ 20%, et il passe pour tirer. Ensuite, nous allons à 100%, revenons à nouveau à 20%, au modèle de traction, etc. Tout cela se produit dans la dynamique, vous n'avez pas besoin d'arrêter le service, entrez quelque chose dans la configuration. Ceci est une illustration du fonctionnement de la contre-pression dans Akka Streams.

Principes d'Akka Streams

Bien sûr, Akka Streams ne gagnerait pas en popularité s'il n'y avait pas de blocs intégrés très faciles à utiliser. Il y en a beaucoup. Ils sont divisés en trois groupes principaux:

1. Source de données (Source) - étape de traitement avec une sortie.
   
2. Le récepteur est une étape de traitement à entrée unique.
   
3. Checkpoint (Flow) - étape de traitement avec une entrée et une sortie. Des transformations fonctionnelles ont lieu ici, et pas nécessairement dans la mémoire: il peut s'agir, par exemple, d'un appel à un service web, à certains éléments de parallélisme, multi-thread.
   

De ces trois types, des graphiques peuvent être formés. Ce sont déjà des étapes de traitement plus complexes, qui sont construites à partir de sources, de drains et de points de contrôle. Mais tous les graphiques ne peuvent pas être exécutés: s'il y a des trous dedans, c'est-à-dire des entrées et des sorties ouvertes, alors ce graphique n'est pas exécuté.
Un graphe est un graphe exécutable, s'il est fermé, c'est-à-dire qu'il y a une sortie pour chaque entrée: si les données sont entrées, elles doivent être allées quelque part.



Akka Streams a des sources intégrées: dans l'image, vous voyez combien d'entre elles. Leurs noms sont à peu près un à un et reflètent ce que Scala ou la JVM a, à l'exception de certaines sources utiles spécifiques à .NET. Les deux premiers (FromEnumerator et From) sont parmi les plus importants: toute numérotation, tout ienumerable peut être transformé en source de flux.



Il existe des drains intégrés: certains d'entre eux ressemblent aux méthodes LINQ, par exemple, First, Last, FirstOrDefault. Bien sûr, tout ce que vous obtenez, vous pouvez le vider dans des fichiers, dans des flux, non pas dans des flux Akka, mais dans des flux .NET. Et encore une fois, si vous avez des acteurs dans votre système, vous pouvez les utiliser à la fois en entrée et en sortie du système, c'est-à-dire, si vous le souhaitez, les intégrer dans votre système fini.



Et il existe un grand nombre de points de contrôle intégrés, qui rappellent peut-être encore plus LINQ, car ici il y a Select, SelectMany et GroupBy, c'est-à-dire tout ce que nous avons l'habitude de travailler avec LINQ.

Par exemple, Select in Scala est appelé SelectAsync: il est suffisamment puissant car il prend le niveau de parallélisme comme l'un des arguments. Autrement dit, vous pouvez indiquer que, par exemple, Select envoie des données à un service Web en parallèle sur dix threads, puis elles sont toutes collectées et transmises. En fait, vous déterminez le degré de mise à l'échelle du point de contrôle avec une ligne de code.

Une déclaration de flux est son plan d'exécution, c'est-à-dire qu'un graphique, même un graphique d'exécution, ne peut pas être exécuté comme ça - il doit être matérialisé. Il doit y avoir un système instancié, un système d'acteur, vous devez lui donner un flux, ce plan d'exécution, puis il sera exécuté. De plus, au moment de l'exécution, il est hautement optimisé, tout comme lorsque vous envoyez une expression LINQ à une base de données: un fournisseur peut optimiser votre SQL pour une sortie de données plus efficace, en remplaçant essentiellement la commande de requête par une autre. Même chose avec Akka Streams: à partir de la version 2.0, vous pouvez définir un certain nombre de points de contrôle, et le système comprendra que certains d'entre eux peuvent être combinés afin qu'ils soient exécutés par un seul acteur (fusion d'opérateur). Les points de contrôle, en règle générale, conservent l'ordre des éléments de traitement.

 var results = db.Companies .Join(db.People, c => c.CompanyID, p => p.PersonID, (c, p) => new { c, p }) .Where(z => zcCreated >= fromDate) .OrderByDescending(z => zcCreated) .Select(z => zp) .ToList(); 

La matérialisation du flux peut être comparée au dernier élément ToList de l'expression LINQ dans l'exemple ci-dessus. Si nous n'écrivons pas ToList, nous obtenons alors une expression LINQ non matérialisée qui n'entraînera pas le transfert des données vers le serveur SQL ou Oracle, car la plupart des fournisseurs LINQ prennent en charge l'exécution dite de requête différée (exécution de requête retardée), t c'est-à-dire que la demande n'est exécutée que lorsqu'une commande est donnée pour donner un résultat. Selon ce qui est demandé - une liste ou le premier résultat - l'équipe la plus efficace sera formée. Lorsque nous disons ToList, nous demandons ainsi au fournisseur LINQ de nous donner le résultat final.

 var runnable = Source .From(Enumerable.Range(1, 1000)) .Via(Flow.Create<int>().Select(x => x * 2) .To(Sink.ForEach<int>(x => Console.Write(x.ToString)); 

Akka Streams fonctionne de la même manière. Dans l'image est notre graphique lancé, qui se compose d'une source de points de contrôle et de ruissellement, et nous voulons maintenant l'exécuter.

 var runnable = Source .From(Enumerable.Range(1, 1000)) .Via(Flow.Create<int>().Select(x => x * 2) .To(Sink.ForEach<int>(x => Console.Write(x.ToString)); var system = ActorSystem.Create("MyActorSystem"); using (var materializer = ActorMaterializer.Create(system)) { await runnable.Run(materializer); } 

Pour que cela se produise, nous devons créer un système d'acteurs, en lui il y a un matérialisant, lui passer notre graphique, et il l'exécutera. Si nous le recréons, il l'exécutera à nouveau et d'autres résultats pourront être obtenus.

En plus de la matérialisation du flux, en parlant de la partie matérielle d'Akka Streams, il convient de mentionner les valeurs matérialisées.

 var output = new List<int>(); var source1 = Source.From(Enumerable.Range(1, 1000)); var sink1 = Sink.ForEach<int>(output.Add); IRunnableGraph<NotUsed> runnable1 = source1.To(sink1); var source2 = Source.From(Enumerable.Range(1, 1000)); var sink2 = Sink.Sum<int>((x,y) => x + y); IRunnableGraph<Task<int>> runnable2 = source2.ToMaterialized(sink2, Keep.Right); 

Lorsque nous avons un flux qui va de la source aux points de contrôle vers le drain, alors si nous ne demandons pas de valeurs intermédiaires, elles ne sont pas disponibles, car elles seront exécutées de la manière la plus efficace. C'est comme une boîte noire. Mais il peut être intéressant pour nous de retirer certaines valeurs intermédiaires, car à chaque point à gauche, certaines valeurs entrent, d'autres valeurs sortent à droite, et vous pouvez spécifier un graphique pour indiquer ce qui vous intéresse. Dans l'exemple ci-dessus, un graphique de départ dans lequel NotUsed est indiqué, c'est-à-dire qu'aucune valeur matérialisée ne nous intéresse. Ci-dessous, nous le créons avec l'indication que sur le côté droit du ruissellement, c'est-à-dire, une fois toutes les transformations terminées, nous devons donner des valeurs matérialisées. Nous obtenons la tâche graphique - une tâche, à la fin de laquelle nous obtenons un int, c'est-à-dire ce qui se passe à la fin de ce graphique. Vous pouvez indiquer dans chaque paragraphe que vous avez besoin d'une sorte de valeurs matérialisées, tout cela sera progressivement collecté.

Pour transférer des données dans les flux Akka Streams ou pour les retirer de là, bien sûr, une sorte d'interaction avec le monde extérieur est nécessaire. Les étages source intégrés contiennent une large gamme de flux de données réactifs:

• Source.FromEnumerator et Source.From vous permettent de transférer des données à partir de n'importe quelle source qui implémente IEnumerable;
  
• Déplier et DéplierAsync génèrent les résultats des calculs de fonction à condition qu'il renvoie des valeurs non nulles;
  
• FromInputStream transforme un flux;
  
• FromFile analyse le contenu du fichier dans le flux réactif;
  
• ActorPublisher convertit les messages des acteurs.
  

Comme je l'ai déjà dit, pour les développeurs .NET, il est très productif d'utiliser Enumerator ou IEnumerable, mais parfois c'est trop primitif, trop inefficace pour accéder aux données. Des sources plus complexes contenant une grande quantité de données nécessitent des connecteurs spéciaux. Ces connecteurs sont écrits. Il existe un projet open source Alpakka, qui est apparu à l'origine dans Scala et est maintenant dans .NET. De plus, Akka a des acteurs dits persistants et ils ont leurs propres flux qui peuvent être utilisés (par exemple, Akka Persistence Query forme le flux de contenu du Akka Event Journal).



Si vous travaillez avec Scala, le moyen le plus simple est pour vous: il existe un grand nombre de connecteurs et vous trouverez sûrement quelque chose à votre goût. Pour information, Kafka est la soi-disant Reactive Kafka, pas Kafka Streams. Pour autant que je sache, Kafka Streams ne supporte pas la contre-pression. Reactive Kafka est une implémentation de flux de Kafka qui prend en charge Reactive Streams.



La liste des connecteurs Alpakka .NET est plus modeste, mais elle est réapprovisionnée et il y a un élément de concurrence. Il y a un tweet de demi-an de David Fowler de Microsoft, qui a déclaré que SignalR peut désormais échanger des données avec Reactive Extensions, et l'un des développeurs d'Akka a répondu qu'il se trouvait en fait dans Akka Streams depuis un certain temps. Akka prend en charge divers services de Microsoft Azure. CSV est le résultat de la frustration d'Aaron Stannard lorsqu'il a découvert qu'il n'y avait pas de bon flux pour CSV: maintenant Akka a son propre flux pour CSV XML. Il y a AMQP (en réalité, RabbitMQ), il est en cours de développement, mais est disponible pour utilisation, ça marche. Kafka est également en cours de développement. Cette liste continuera de s'étendre.

Quelques mots sur les alternatives, car si vous travaillez avec des flux de données, Akka Streams n'est bien sûr pas le seul moyen de gérer ces flux. Très probablement, dans votre projet, le choix de la façon d'implémenter les threads dépendra de nombreux autres facteurs qui peuvent devenir essentiels. Par exemple, si vous travaillez beaucoup avec Microsoft Azure et Orléans est organiquement intégré aux besoins de votre projet avec leur prise en charge des acteurs virtuels, ou, comme ils les appellent, grains, alors ils ont leur propre implémentation qui ne répond pas à la spécification Reactive Streams - Orleans Streams, qui il sera le plus proche de vous et il est logique que vous y prêtiez attention. Si vous travaillez beaucoup avec TPL, il y a TPL DataFlow - cela peut être l'analogie la plus proche des flux Akka: il existe également des primitives pour lier les flux de données, ainsi que des outils de mise en mémoire tampon et de limitation de la bande passante (BoundedCapacity, MaxMessagePerTask). Si les idées du modèle d'acteur sont proches de vous, alors Akka Streams est un moyen de résoudre ce problème et de gagner beaucoup de temps sans avoir à écrire chaque acteur manuellement.

Exemple d'implémentation: flux du journal des événements

Regardons quelques exemples d'implémentation. — , . Akka Streams, , - , .



. : 15 23 , 7 . — . Kibana Dashboard.
Kibana Elasticsearch , Elasticsearch , , , , . , , , . . . (event journal) Akka, Microsoft SQL Server. , .

 CREATE TABLE EventJournal ( Ordering BIGINT IDENTITY(1,1) PRIMARY KEY NOT NULL, PersistenceID NVARCHAR(255) NOT NULL, SequenceNr BIGINT NOT NULL, Timestamp BIGINT NOT NULL, IsDeleted BIT NOT NULL, Manifest NVARCHAR(500) NOT NULL, Payload VARBINARY(MAX) NOT NULL, Tags NVARCHAR(100) NULL CONSTRAINT QU_EventJournal UNIQUE (PersistenceID, SequenceNr) ) 

, , , , SQL Server, eventstore Akka, eventJournal. eventstore.



, . , , , , - : , . , . . . - . , . , . , Akka persistence query.



, , .

(persistence queries):

• AllPersistencelds
• CurrentPersistencelds
• EventsByPersistenceld
• CurrentEventsByPersistenceld
• EventsByTag
• CurrentEventsByTag

, , , Current — , . — . EventsByTag.

 let system = mailbox.Context.System let queries = PersistenceQuery.Get(system) .ReadJournalFor<SqlReadJournal>(SqlReadJournal.Identifier) let mat = ActorMaterializer.Create(system) let offset = getCurrentOffset client config let ks = KillSwitches.Shared "persistence-elastic" let task = queries.EventsByTag(PersistenceUtils.anyEventTag, offset) .Select(fun e -> ElasticTypes.EventEnvelope.FromAkka e) .GroupedWithin(config.BatchSize, config.BatchTimeout) .Via(ks.Flow()) .RunForeach((fun batch -> processItems client batch), mat) .ContinueWith(handleStreamError mailbox, TaskContinuationOptions.OnlyOnFaulted) |> Async.AwaitTaskVoid 

, . F#, C# . EventsByTag, Akka Streams, , Elasticsearch. . . - , , , — . .

. , , , , Twitter , — , , , . , Akka Streams.

:

Akka Scala, Akka.NET, , , , , . . - . Tweetinvi — , Twitter, . Reactive Streams, . . , , , , - Akka, , .



, , . . Broadcast-. , , . : , , , , .

GitHub-, AkkaStreamsDemo . ( ).

Commençons par un simple. Twitter: Program.cs

 var useCachedTweets = false 

, Twitter, , . RunnableGraph.

 public static IRunnableGraph<IActorRef> CreateRunnableGraph() { var tweetSource = Source.ActorRef<ITweet>(100, OverflowStrategy.DropHead); var formatFlow = Flow.Create<ITweet>().Select(Utils.FormatTweet); var writeSink = Sink.ForEach<string>(Console.WriteLine); return tweetSource.Via(formatFlow).To(writeSink); } 

( )

, . , , ( ) .

StartTweetStream — Tweetinvi.

 public static void StartTweetStream(IActorRef actor) { var stream = Stream.CreateSampleStream(); stream.TweetReceived += (_, arg) => { arg.Tweet.Text = arg.Tweet.Text.Replace("\r", " ").Replace("\n", " "); var json = JsonConvert.SerializeObject(arg.Tweet); File.AppendAllText("tweets.txt", $"{json}\r\n"); actor.Tell(arg.Tweet); }; stream.StartStream(); } 

( )

CreateSampleStream , . , , , : « ». IEnumerable, .

TweetEnumerator : , Current, MoveNext, Reset, Dispose, . , . , . .

useCachedTweets true, . CashedTweets — , 50000 , , , . , , . — . , .

TweetsWithBroadcast:

 var graph = GraphDsl.Create(b => { var broadcast = b.Add(new Broadcast<ITweet>(2)); var merge = b.Add(new Merge<string>(2)); b.From(broadcast.Out(0)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.CreatedBy)) .Via(formatUser) .To(merge.In(0)); b.From(broadcast.Out(1)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.Coordinates)) .Via(formatCoordinates) .To(merge.In(1)); return new FlowShape<ITweet, string>(broadcast.In, merge.Out); }); 

( )

Scala, , DSL. Broadcast — out(0), out(1) — CreatedBy, , . .

— . .

 var graph = GraphDsl.Create(b => { var broadcast = b.Add(new Broadcast<ITweet>(2)); var merge = b.Add(new Merge<string>(2)); b.From(broadcast.Out(0)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.CreatedBy) .Throttle(10, TimeSpan.FromSeconds(1), 1, ThrottleMode.Shaping)) .Via(formatUser) .To(merge.In(0)); b.From(broadcast.Out(1)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.Coordinates) .Buffer(10, OverflowStrategy.DropNew) .Throttle(1, TimeSpan.FromSeconds(1), 10, ThrottleMode.Shaping)) .Via(formatCoordinates) .To(merge.In(1)); return new FlowShape<ITweet, string>(broadcast.In, merge.Out); });} 

( )

10 , 10. , , , . , , Akka Streams Reactive Streams: . , , , , - . , , , . , . , , . Buffer(10, OverFlowStrategy.DropHead). , . 10 , . , , - , — - , , , , . . . , .

 var graph = GraphDsl.Create(b => { var broadcast = b.Add(new Broadcast<ITweet>(2)); var merge = b.Add(new Merge<string>(2)); b.From(broadcast.Out(0)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.CreatedBy) .Throttle(10, TimeSpan.FromSeconds(1), 1, ThrottleMode.Shaping)) .Via(formatUser) .To(merge.In(0)); b.From(broadcast.Out(1)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.Coordinates) .Buffer(10, OverflowStrategy.DropNew) .Throttle(1, TimeSpan.FromSeconds(1), 10, ThrottleMode.Shaping)) .Via(Flow.Create<ICoordinates>().SelectAsync(5, Utils.GetWeatherAsync)) .Via(formatTemperature) .To(merge.In(1)); return new FlowShape<ITweet, string>(broadcast.In, merge.Out); }); 

( )

, SelectAsync, . , , 5: , 5 , , . , , .

 public static async Task<decimal> GetWeatherAsync(ICoordinates coordinates) { var httpClient = new HttpClient(); var requestUrl = $"http://api.met.no/weatherapi/locationforecast/1.9/?lat={coordinates.Latitude};lon={coordinates.Latitude}"; var result = await httpClient.GetStringAsync(requestUrl); var doc = XDocument.Parse(result); var temp = doc.Root.Descendants("temperature").First().Attribute("value").Value; return decimal.Parse(temp); } 

( )

. -, , - , HttpClient , XML, , .

, , , . 10 10 , , .

, — , . , Akka Streams, , . , , .

, , , Akka Streams, . , , Akka Streams, C# , , , , .

Akka Streams , ? DotNext 2017 Moscow Azure Functions. - , deployment, , ( - , , ), . , , , . , , Akka Streams, .. , . .

Akka Streams , , , , , . , , , , , . Akka Streams — , , .

, Akka Streams, «Akka Stream Rap». , .

This is the Akka Stream.

This is the Source that feeds the Akka Stream.

This is the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Streams.

This is the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Drop that removes from the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is TakeWhile that pulls from the Drop that removes from the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Throttle that speeds down the TakeWhile that pulls from the Drop that removes from the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Bidiflow that turns back the Throttle that speeds down the TakeWhile that pulls from the Drop that removes from the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the source that feeds the Akka Streams.

This is the Sink that is filled from the Bidiflow that turns back the Throttle that speeds down the TakeWhile that pulls from the Drop that removes from the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

. — 22-23 DotNext 2018 Moscow , . ( ).