Hola En este artículo, galoparé por Europa, es decir, te diré lo que quieren decir con programación reactiva, te presentaré a actores, flujos reactivos y, finalmente, utilizando flujos reactivos, reconoceremos los gestos del mouse, como en la antigua Opera y su sucesor espiritual: Vivaldi .

El objetivo es presentar los conceptos básicos de la programación reactiva y mostrar que no todo es tan complicado y aterrador como podría parecer a primera vista.

imagen

Fuente

¿Qué es la programación reactiva?

Para responder a esta pregunta, nos dirigimos al sitio . Tiene una hermosa imagen que muestra 4 criterios principales que deben cumplir las aplicaciones reactivas.

imagen

La aplicación debe ser rápida, tolerante a fallas y escalar bien.
Parece que "somos para todos los buenos contra todos los malos", ¿verdad?

Qué se entiende por estas palabras:

Capacidad de respuesta

La aplicación debe dar al usuario el resultado en medio segundo. Esto también incluye el principio de falla rápida: es decir, cuando algo sale mal, es mejor devolver un mensaje de error como “Lo siento, hubo un problema. Intente nuevamente más tarde que hacer que el clima espere junto al mar. Si la operación es larga, le mostramos al usuario una barra de progreso. Si es muy largo: “su solicitud se cumplirá tentativamente el 18 de marzo de 2042. Le enviaremos una notificación por correo ".
La escalabilidad es una forma de proporcionar capacidad de respuesta bajo carga. Imagine el ciclo de vida de un servicio relativamente exitoso:
1. Lanzamiento: el flujo de solicitud es pequeño, el servicio se ejecuta en una máquina virtual con un núcleo.
2. El flujo de solicitudes aumenta: los núcleos se agregan a la máquina virtual y las solicitudes se procesan en varios subprocesos.
3. Aún más carga, conectamos por lotes, las solicitudes a la base de datos y al disco duro se agrupan.
4. Aún más carga: debe generar más servidores y proporcionar trabajo en el clúster.
  Idealmente, el sistema en sí debería aumentar o disminuir según la carga.
Tolerancia a fallos

Aceptamos que vivimos en un mundo imperfecto y que todo sucede. En caso de que algo salga mal en nuestro sistema, debemos proporcionar métodos de manejo y recuperación de errores
Y finalmente, estamos invitados a lograr todo esto usando un sistema cuya arquitectura se base en mensajes basados en mensajes

Antes de continuar, quiero detenerme en cómo los sistemas basados en eventos difieren de los sistemas basados en mensajes.

Impulsado por eventos:

Evento: el sistema informa que ha alcanzado un cierto estado.
Puede haber muchos suscriptores al evento.
La cadena de eventos suele ser corta y los controladores de eventos están cerca (tanto físicamente como en código) de la fuente.
El origen del evento y sus controladores generalmente tienen un estado común (físicamente, usan la misma porción de RAM para el intercambio de información).

A diferencia de los eventos, en un sistema de mensajes:

Cada mensaje tiene un solo destinatario.
Los mensajes son inmutables: no puede cambiar nada en el mensaje recibido para que el remitente lo sepa y pueda leer la información.
Los elementos del sistema responden (o no responden) a la recepción de mensajes y pueden enviar mensajes a otros elementos del sistema.

Todo esto nos ofrece

Modelo de actor

Hitos del desarrollo:

La primera mención de actores está en un artículo científico de 1973: Carl Hewitt, Peter Bishop y Richard Steiger, "Un formalismo ACTOR modular universal para la inteligencia artificial".
1986 - Aparece Erlang. Ericson necesitaba un lenguaje para equipos de telecomunicaciones que proporcionara tolerancia a fallas y propagación sin errores. En el contexto de este artículo, sus características principales son:
- Todo es un proceso
- Los mensajes son la única forma de comunicación (Erlang es un lenguaje funcional, y los mensajes en él no se pueden cambiar).
..
2004 - la primera versión del lenguaje Scala. Sus características:
- Desarrollado por JVM,
- Funcional
- Para subprocesos múltiples, se ha seleccionado un modelo de actor.
2009 - la implementación de los actores fue asignada en una biblioteca separada - Akka
2014 - Akka.net - fue portado a .Net.

¿Qué pueden hacer los actores?

Los actores son los mismos objetos, pero:

A diferencia de los objetos ordinarios, los actores no pueden llamar los métodos de los demás.
Los actores pueden transmitir información solo a través de mensajes inmutables .
Al recibir el mensaje, el actor puede
- Crear nuevos actores (serán más bajos en la jerarquía),
- Enviar mensajes a otros actores,
- Detén a los actores de abajo en la jerarquía y a ti mismo.

Veamos un ejemplo.

imagen

El actor A quiere enviar un mensaje al actor B. Todo lo que tiene es ActorRef (alguna dirección). El actor B puede estar en cualquier lugar.
El actor A envía una letra B a través del sistema (ActorSystem). El sistema coloca la carta en el buzón del actor B y "despierta" al actor B. El actor B toma la carta del buzón y hace algo.

En comparación con los métodos de llamada en otro objeto, parece innecesariamente complicado, pero el modelo de actores se adapta perfectamente al mundo real, si imagina que los actores son personas que están capacitadas para hacer algo en respuesta a ciertos estímulos.

Imagina un padre y un hijo:

El padre envía a su hijo SMSku "Limpio en la habitación" y continúa haciendo lo suyo. El hijo lee SMSku y comienza a limpiar. Padre, mientras tanto, está jugando al póker. El hijo termina de limpiar y envía un mensaje de texto "Finalizar". Se ve simple, ¿verdad?

Ahora imagine que padre e hijo no son actores, sino objetos ordinarios que pueden extraer los métodos del otro. El padre tira de su hijo para el método de "limpiar la habitación" y lo sigue, esperando hasta que el hijo termine de limpiar y transfiera el control a su padre. Padre no puede jugar póker en este momento. En este contexto, el modelo de actor se está volviendo más atractivo.

Ahora pasemos a

Akka.NET

Todo lo que está escrito a continuación es cierto para el Akka original para la JVM, pero para mí, C # está más cerca que Java, por lo que utilizaré Akka.NET como ejemplo.

¿Cuáles son los beneficios de Akka?

Multithreading a través de mensajes. Ya no tiene que sufrir con todo tipo de bloqueos, semáforos, mutexes y otros encantos característicos de los subprocesos múltiples clásicos con memoria compartida.
Comunicación transparente entre el sistema y sus componentes. No necesita preocuparse por el código de red complejo: el sistema mismo encontrará el destino del mensaje y garantizará la entrega del mensaje (aquí puede insertar un chiste sobre UDP vs TCP).
Arquitectura flexible que puede escalar automáticamente hacia arriba o hacia abajo. Por ejemplo, bajo carga, el sistema puede generar nodos de clúster adicionales y distribuir la carga de manera uniforme.

Pero el tema del escalado es muy extenso y merece una publicación separada. Por lo tanto, contaré con más detalle solo sobre la función, que será útil en todos los proyectos:

Manejo de errores

Los actores tienen una jerarquía: se puede representar como un árbol. Cada actor tiene un padre y puede tener "hijos".

imagen

Documentación de Akka.NET Copyright 2013-2018 Proyecto Akka.NET

Para cada actor, puede establecer una estrategia de supervisión: qué hacer si algo sale mal para los "niños". Por ejemplo, "vencer" a un actor que tiene problemas, y luego crear un nuevo actor del mismo tipo y confiarle el mismo trabajo.

Por ejemplo, hice una aplicación en Akka.net CRUD, en la que la capa de "lógica de negocios" se implementa en los actores. El objetivo de este proyecto era descubrir si los actores deberían usarse en sistemas no escalables: ¿mejorarán la vida o añadirán más dolor?

Cómo puede ayudar el manejo de errores incorporado de Akka:

GIF

todo está bien, la aplicación funciona,
Algo le sucedió al repositorio, y ahora da el resultado solo 1 de cada 5 veces,
Establecí la estrategia de supervisión para "probar 10 veces por segundo",
la aplicación vuelve a funcionar (aunque es más lenta) y tengo tiempo para averiguar qué sucede.

Existe la tentación de decir: "Vamos, escribiré ese error manejando yo mismo, ¿por qué algunos actores tienen que cometer un error?" Comentario justo, pero solo si los puntos de falla son pocos.

Y algo de código. Así es como se ve la inicialización del sistema de actores en el contenedor IoC:

public Container() { system = ActorSystem.Create("MySystem"); var echo = system.ActorOf<EchoActor>("Echo"); //stop initialization if something is wrong with actor system var alive = echo.Ask<bool>(true, TimeSpan.FromMilliseconds(100)).Result; container = new WindsorContainer(); //search for dependencies //register controllers //register ActorSystem propsResolver = new WindsorDependencyResolver(container, (ActorSystem)system); system.AddDependencyResolver(propsResolver); actorSystemWrapper = new ActorSystemWrapper(system, propsResolver); container.Register(Component.For<IActorRefFactory>().Instance(actorSystemWrapper)); container.Register(Component.For<IDependencyResolver>().Instance(propsResolver)); }

EchoActor es el actor más simple que devuelve un valor al remitente:

  public class EchoActor : ReceiveActor { public EchoActor() { Receive<bool>(flag => { Sender.Tell(flag); }); } }

Para conectar a los actores con el código "regular", se utiliza el comando Preguntar:

  public async Task<ActionResult> Index() { ViewBag.Type = typeof(Model); var res = await CrudActorRef.Ask<IEnumerable<Model>>(DataMessage.GetAll<Model>(), maxDelay); return View(res); }

Total

Riéndome con los actores, puedo decir:

Míralos si necesitas escalabilidad.
Para una lógica empresarial compleja, es mejor no usarlos debido a
- Inyección de dependencia extraña. Para inicializar un actor con las dependencias necesarias, primero debe crear un objeto Props y luego entregarlo al ActorSystem para crear un actor del tipo deseado. Para crear accesorios utilizando contenedores IoC (por ejemplo, Castle Windsor o Autofac) hay envoltorios listos para usar: DependencyResolvers. Pero me enfrenté al hecho de que el contenedor IoC estaba tratando de controlar la vida útil de la dependencia, y después de un tiempo el sistema se cayó silenciosamente.
  
  * Quizás, en lugar de inyectar una dependencia en un objeto, debería colocar esta dependencia como un actor secundario.
- Problemas de mecanografía. ActorRef no sabe nada sobre el tipo de actor al que se refiere. Es decir, en tiempo de compilación no se sabe si un actor puede procesar un mensaje de este tipo o no.

Parte 2: corrientes en chorro

Ahora pasemos a un tema más popular y útil: los flujos de chorro. Si nunca puede reunirse con actores en el proceso de trabajo, entonces las transmisiones de Rx serán útiles tanto en la interfaz como en el backend. Su implementación está en casi todos los lenguajes de programación modernos. Daré ejemplos sobre RxJs, porque hoy en día incluso los programadores de backend a veces tienen que hacer algo en JavaScript.

Las transmisiones Rx están disponibles para todos los lenguajes de programación populares.

" Introducción a la programación reactiva que te has estado perdiendo " por Andre Staltz , licenciado bajo CC BY-NC 4.0

Para explicar qué es la corriente en chorro, comenzaré con las colecciones pull y push.

	Valor de retorno único	Múltiples valores de retorno
Tirar Síncrono Interactivo	T	IEnumerable <T>
Empujar Asincrónico Reactivo	Tarea <T>	IObservable <T>

Las colecciones pull son a lo que todos estamos acostumbrados en la programación. El ejemplo más llamativo es una matriz.

 const arr = [1,2,3,4,5];

Ya tiene datos, él mismo no cambiará estos datos, pero puede proporcionarlos a pedido.

 arr.forEach(console.log);

Además, antes de hacer algo con los datos, puede procesarlos de alguna manera.

 arr.map(i => i+1).map(I => “my number is ”+i).forEach(console.log);

Ahora imaginemos que inicialmente no hay datos en la colección, pero definitivamente le informará que han aparecido (Push). Y al mismo tiempo, aún podemos aplicar las transformaciones necesarias a esta colección.

Por ejemplo:

 source.map(i => i+1).map(I => “my number is ”+i).forEach(console.log);

Cuando aparece un valor como 1 en la fuente, console.log generará "mi número es 1".

Cómo funciona

Aparece una nueva entidad: Asunto (u Observable):

 const observable = Rx.Observable.create(function (observer) { observer.next(1); observer.next(2); observer.next(3); setTimeout(() => { observer.next(4); observer.complete(); }, 1000); });

Esta es una colección push que enviará notificaciones sobre cambios en su estado.

En este caso, los números 1, 2 y 3 aparecerán inmediatamente en él, en un segundo 4, y luego la colección "terminará". Este es un tipo de evento tan especial.

La segunda entidad es Observador. Puede suscribirse a los eventos del Asunto y hacer algo con los datos recibidos. Por ejemplo:

 observable.subscribe(x => console.log(x)); observable.subscribe({ next: x => console.log('got value ' + x), error: err => console.error('something wrong occurred: ' + err), complete: () => console.log('done'), }); observable .map(x => 'This is ' + x) .subscribe(x => console.log(x));

Se puede ver que un sujeto puede tener muchos suscriptores.

Parece fácil, pero aún no está claro por qué es necesario. Le daré 2 definiciones más que necesita saber cuando trabaje con flujos reactivos, y luego le mostraré en la práctica cómo funcionan y en qué situaciones se revela todo su potencial.

Observables en frío

Notificar sobre eventos cuando alguien se suscribe a ellos.
El flujo de datos completo se envía nuevamente a cada suscriptor, independientemente del momento de la suscripción.
Los datos se copian para cada suscriptor.

¿Qué significa esto? Digamos que la empresa (Asunto) decidió organizar la distribución de regalos. Cada empleado (Observador) viene al trabajo y recibe su copia del regalo. Nadie queda privado.

Observables calientes

Intentan notificar el evento independientemente de la presencia de suscriptores. Si en el momento del evento no había suscriptores, los datos se pierden.

Ejemplo: en la mañana, se traen pasteles calientes para los empleados a la empresa. Cuando los traen, todas las alondras vuelan al olor y preparan los pasteles para el desayuno. Pero los búhos que vinieron después ya no reciben pasteles.

¿En qué situaciones se utilizan las corrientes en chorro?

Cuando hay un flujo de datos distribuido en el tiempo. Por ejemplo, entrada del usuario. O registros de cualquier servicio. En uno de los proyectos, vi un registrador hecho a sí mismo que recopilaba eventos en N segundos y luego grababa simultáneamente todo el paquete. El código de la batería ocupaba la página. Si se usaran transmisiones Rx, entonces sería mucho más simple:

imagen

“ RxJs Reference / Observable , documentación bajo licencia CC BY 4.0 .
(hay muchos ejemplos e imágenes que explican lo que hacen varias operaciones con flujos reactivos)

 source.bufferTime(2000).subsribe(doThings);

Y finalmente, un ejemplo de uso.

Reconociendo gestos del mouse con transmisiones Rx

En la antigua Opera o su sucesor espiritual, Vivaldi, había un control de navegador que usaba gestos del mouse.

Gif - gestos del mouse en Vivaldi

Es decir, debe reconocer los movimientos del mouse hacia arriba / abajo, derecha / izquierda y sus combinaciones. Se puede escribir sin secuencias Rx, pero el código será complejo y difícil de mantener.

Y así es como se ve con las transmisiones Rx:

Comenzaré desde el final: estableceré qué datos y en qué formato buscaré en la secuencia original:

 //gestures to look for const gestures = Rx.Observable.from([ { name: "Left", sequence: Rx.Observable.from([{ x: -1, y: 0 }]) }, { name: "Right", sequence: Rx.Observable.from([{ x: 1, y: 0 }]) }, { name: "Up", sequence: Rx.Observable.from([{ x: 0, y: -1 }]) }, { name: "Down", sequence: Rx.Observable.from([{ x: 0, y: 1 }]) }, { name: "Down+Up", sequence: Rx.Observable.from([{ x: 0, y: 1 }, { x: 0, y: -1 }]) }, { name: "Up+Right", sequence: Rx.Observable.from([{ x: 0, y: -1 }, { x: 1, y: 0 }]) } ]);

Estos son vectores unitarios y sus combinaciones.

A continuación, debe convertir los eventos del mouse en secuencias Rx. Todas las bibliotecas Rx tienen herramientas integradas para convertir eventos estándar en Observables.

 const mouseMoves = Rx.Observable.fromEvent(canvas, 'mousemove'), mouseDowns = Rx.Observable.fromEvent(canvas, 'mousedown'), mouseUps = Rx.Observable.fromEvent(canvas, 'mouseup');

Luego, agrupo las coordenadas del mouse por 2 y encuentro su diferencia, obteniendo el desplazamiento del mouse.

 const mouseDiffs = mouseMoves .map(getOffset) .pairwise() .map(pair => { return { x: pair[1].x-pair[0].x, y: pair[1].y-pair[0].y } });

Y agrupe estos movimientos usando los eventos 'mousedown' y 'mouseup'.

 const mouseGestures = mouseDiffs .bufferToggle(mouseDowns, x => mouseUps) .map(concat);

La función concat corta los movimientos que son demasiado cortos y agrupa los movimientos que están más o menos alineados en dirección.

 function concat(values) {//summarize move in same direction return values.reduce((a, v) => { if (!a.length) { a.push(v); } else { const last = a[a.length - 1]; const lastAngle = Math.atan2(last.x, last.y); const angle = Math.atan2(vx, vy); const angleDiff = normalizeAngle(angle - lastAngle); const dist = Math.hypot(vx, vy); if (dist < 1) return a;//move is too short – ignore //moving in same direction => adding vectors if (Math.abs(angleDiff) <= maxAngleDiff) { last.x += vx; last.y += vy; } else { a.push(v); } } return a; }, []); }

Si el movimiento en el eje X o Y es demasiado corto, se restablece a cero. Y entonces solo queda el signo de las coordenadas de desplazamiento obtenidas. Por lo tanto, se obtienen los vectores unitarios que estábamos buscando.

 const normalizedMouseGestures = mouseGestures.map(arr => arr.map(v => { const dist = Math.hypot(vx, vy);//length of vector vx = Math.abs(vx) > minMove && Math.abs(vx) * treshold > dist ? vx : 0; vy = Math.abs(vy) > minMove && Math.abs(vy) * treshold > dist ? vy : 0; return v; }) ).map(arr => arr .map(v => { return { x: Math.sign(vx), y: Math.sign(vy) }; }) .filter(v => Math.hypot(vx, vy) > 0) );

Resultado:

 gestures.map(gesture => normalizedMouseGestures.mergeMap( moves => Rx.Observable.from(moves) .sequenceEqual(gesture.sequence, comparer) ).filter(x => x).mapTo(gesture.name) ).mergeAll().subscribe(gestureName => actions[gestureName]());

Usando secuenciaEqual, puede comparar los movimientos recibidos con los originales y, si hay una coincidencia, realizar una determinada acción.