¿Cómo las corrutinas de Kotlin son mejores que las de RxKotlin?

Solución:

Descargo de responsabilidad: Partes de esta respuesta son irrelevantes ya que Coroutines ahora tienen la API de flujo, muy similar a Rx one. Si desea una respuesta actualizada, salte a la última edición.

Hay dos partes en Rx; el patrón Observable, y un conjunto sólido de operadores para manipularlos, transformarlos y combinarlos. El patrón observable, por sí solo, no hace mucho. Lo mismo con Coroutines; es solo otro paradigma para lidiar con el asincronismo. Puede comparar los pros y los contras de las devoluciones de llamada, Observable y Coroutines para resolver un problema dado, pero no puede comparar un paradigma con una biblioteca con todas las funciones. Es como comparar un lenguaje con un marco.

¿Cómo las corrutinas de Kotlin son mejores que las de RxKotlin? Todavía no usé corrutinas, pero se parece a async / wait en C #. Simplemente escribe código secuencial, todo es tan fácil como escribir código síncrono … excepto que se ejecuta de forma asincrónica. Es más fácil de comprender.

¿Por qué querría usar corutinas de Kotlin? Yo responderé por mí mismo. La mayoría de las veces me ceñiré a Rx, porque prefiero la arquitectura impulsada por eventos. Pero si surgiera la situación en la que estoy escribiendo código secuencial y necesito llamar a un método asincrónico en el medio, felizmente aprovecharé las corrutinas para mantenerlo así y evitar envolver todo en Observable.

Editar: Ahora que estoy usando corrutinas, es hora de una actualización.

RxKotlin es simplemente azúcar sintáctico para usar RxJava en Kotlin, así que hablaré sobre RxJava y no sobre RxKotlin a continuación. Las corrutinas son una palanca más baja y un concepto más general que RxJava, sirven a otros casos de uso. Dicho esto, hay un caso de uso en el que puede comparar RxJava y corrutinas (channel), está transmitiendo datos de forma asincrónica. Las corrutinas tienen una clara ventaja sobre RxJava aquí:

Las corrutinas son mejores para lidiar con los recursos

• En RxJava puede asignar cálculos a los programadores pero subscribeOn() y ObserveOn()son confusos. A cada corrutina se le da un contexto de hilo y regresa al contexto principal. Para un canal, ambos lados (productor, consumidor) ejecutan en su propio contexto. Las corrutinas son más intuitivas en la afectación de subprocesos o grupos de subprocesos.
• Las corrutinas dan más control sobre cuándo ocurren esos cálculos. Por ejemplo, puede pasar la mano (yield), priorizar (select), paralelizar (múltiples producer/actor sobre channel) o bloquear recurso (Mutex) para un cálculo dado. Puede que no importe en el servidor (donde RxJava llegó primero) pero en un entorno con recursos limitados, este nivel de control puede ser necesario.
• Debido a su naturaleza reactiva, la contrapresión no encaja bien en RxJava. En el otro extremo send() to channel es una función suspensiva que se suspende cuando se alcanza la capacidad del canal. Es una contrapresión inmediata dada por la naturaleza. Tu también podrías offer() al canal, en cuyo caso la llamada nunca se suspende sino que regresa false en caso de que el canal esté lleno, se reproduce de manera efectiva onBackpressureDrop() de RxJava. O simplemente podría escribir su propia lógica de contrapresión personalizada, lo que no será difícil con las corrutinas, especialmente en comparación con hacer lo mismo con RxJava.

Hay otro caso de uso, donde las corrutinas brillan y esto responderá a su segunda pregunta “¿Por qué querría usar las corrutinas de Kotlin?”. Las corrutinas son el reemplazo perfecto para subprocesos de fondo o AsyncTask (Androide). Es tan fácil como launch someBlockingFunction() . Por supuesto, también podría lograr esto con RxJava, usando Schedulers y Completable quizás. No usará (o poco) el patrón Observer y los operadores que son la firma de RxJava, una pista de que este trabajo está fuera del alcance de RxJava. La complejidad de RxJava (un impuesto inútil aquí) hará que su código sea más detallado y menos limpio que la versión de Coroutine.

La legibilidad importa. En este sentido, el enfoque de RxJava y las corrutinas difieren mucho. Las corrutinas son más simples que RxJava. Si no se siente cómodo con map(), flatmap() y programación reactiva funcional en general, las manipulaciones de corrutinas son más fáciles, involucrando instrucciones básicas: for, if, try/catch … Pero personalmente encuentro el código de coroutine más difícil de entender para tareas no triviales. Especialmente implica más anidación y sangría, mientras que el encadenamiento de operadores en RxJava mantiene todo en línea. La programación de estilo funcional hace que el procesamiento sea más explícito. Además de eso, RxJava puede resolver transformaciones complejas con algunos operadores estándar de su rico (OK, demasiado rico) conjunto de operadores. RxJava brilla cuando tiene flujos de datos complejos que requieren muchas combinaciones y transformaciones.

Espero que esas consideraciones le ayuden a elegir la herramienta adecuada según sus necesidades.

Editar: Coroutine ahora tiene flujo, una API muy, muy similar a Rx. Se podrían comparar los pros y los contras de cada uno, pero la verdad es que las diferencias son menores.

Coroutines, como su núcleo, es un patrón de diseño de concurrencia, con bibliotecas complementarias, una de las cuales es una API de flujo similar a Rx. Obviamente, como Coroutines tiene un alcance mucho más amplio que Rx, hay muchas cosas que Coroutines puede que Rx no puede, y no puedo enumerarlas todas. Pero por lo general, si uso Coroutines en uno de mis proyectos, se reduce a una razón:

Las corrutinas son mejores para eliminar la devolución de llamada de su código

Evito usar la devolución de llamada que daña demasiado la legibilidad. Las corrutinas hacen que el código asincrónico sea simple y fácil de escribir. Al aprovechar la palabra clave suspend, su código parece sincrónico.

He visto que Rx se usa en el proyecto principalmente con el mismo propósito de reemplazar la devolución de llamada, pero si no planea modificar su arquitectura para comprometerse con el patrón reactivo, Rx será una carga. Considere esta interfaz:

interface Foo 
   fun bar(callback: Callback)

El equivalente de Coroutine es más explícito, con un tipo de retorno y la palabra clave suspender indicando que es una operación asincrónica.

interface Foo 
   suspend fun bar: Result

Pero hay un problema con el equivalente de Rx:

interface Foo 
   fun bar: Single

Cuando llama a bar () en la devolución de llamada o en la versión Coroutine, activa el cálculo; con la versión Rx, obtiene una representación de un cálculo que puede activar a voluntad. Debe llamar a bar () y luego suscribirse al Single. Por lo general, no es un gran problema, pero es un poco confuso para los principiantes y puede provocar problemas sutiles.

Un ejemplo de tales problemas, suponga que la función de la barra de devolución de llamada se implementa como tal:

fun bar(callback: Callback) 
   setCallback(callback)
   refreshData()

Si no lo transfiere correctamente, terminará con un Single que se puede activar solo una vez porque se llama a refreshData () en la función bar () y no en el momento de la suscripción. Un error de principiante, por supuesto, pero la cosa es que Rx es mucho más que un reemplazo de devolución de llamada y muchos desarrolladores luchan por comprender Rx.

Si su objetivo es transformar una tarea asincrónica de devolución de llamada a un paradigma más agradable, las Coroutines son un ajuste perfecto, mientras que Rx agrega algo de complejidad.

Las corrutinas de Kotlin son diferentes de las de Rx. Es difícil compararlos, porque las corutinas de Kotlin son una característica del lenguaje delgado (con solo un par de conceptos básicos y algunas funciones básicas para manipularlos), mientras que Rx es una biblioteca bastante pesada con una gran variedad de operadores listos para usar. Ambos están diseñados para abordar un problema de programación asincrónica, sin embargo, su enfoque de solución es muy diferente:

• Rx viene con un estilo funcional particular de programación que se puede implementar en prácticamente cualquier lenguaje de programación sin el soporte del propio lenguaje. Funciona bien cuando el problema en cuestión se descompone fácilmente en una secuencia de operadores estándar y no tan bien en caso contrario.

• Las corrutinas de Kotlin proporcionan una función de lenguaje que permite a los escritores de bibliotecas implementar varios estilos de programación asincrónica, que incluyen, entre otros, el estilo reactivo funcional (Rx). Con las corrutinas de Kotlin también puede escribir su código asincrónico en estilo imperativo, en estilo basado en promesas / futuros, en estilo actor, etc.

Es más apropiado comparar Rx con algunas bibliotecas específicas que se implementan en función de las corrutinas de Kotlin.

Tome la biblioteca kotlinx.coroutines como ejemplo. Esta biblioteca proporciona un conjunto de primitivas como async/await y canales que normalmente se incorporan a otros lenguajes de programación. También tiene soporte para actores livianos sin futuro. Puede leer más en la Guía de kotlinx.coroutines por ejemplo.

Canales proporcionados por kotlinx.coroutines puede reemplazar o aumentar Rx en ciertos casos de uso. Hay una guía separada para corrientes reactivas con corrutinas que profundiza en las similitudes y diferencias con Rx.

Conozco muy bien RxJava y recientemente me cambié a Kotlin Coroutines y Flow.

RxKotlin es básicamente lo mismo que RxJava, solo agrega algo de azúcar sintáctico para que sea más cómodo / idiomático escribir código RxJava en Kotlin.

Una comparación “justa” entre RxJava y Kotlin Coroutines debería incluir Flow en la mezcla y trataré de explicar por qué aquí. Esto va a ser un poco largo, pero intentaré que sea lo más simple posible con ejemplos.

Con RxJava tienes diferentes objetos (desde la versión 2):

// 0-n events without backpressure management
fun observeEventsA(): Observable

// 0-n events with explicit backpressure management
fun observeEventsB(): Flowable

// exactly 1 event
fun encrypt(original: String): Single

// 0-1 events
fun cached(key: String): Maybe

// just completes with no specific results
fun syncPending(): Completable

En kotlin coroutines + flow, no necesita muchas entidades porque si no tiene un flujo de eventos, simplemente puede usar corutinas simples (funciones de suspensión):

// 0-n events, the backpressure is automatically taken care off
fun observeEvents(): Flow

// exactly 1 event
suspend fun encrypt(original: String): String

// 0-1 events
suspend fun cached(key: String): MyData?

// just completes with no specific results
suspend fun syncPending()

Bonificación: compatibilidad con Kotlin Flow / Coroutines null valores (soporte eliminado con RxJava 2)

¿Y los operadores?

Con RxJava tienes tantos operadores (map, filter, flatMap, switchMap, …), y para la mayoría de ellos hay una versión para cada tipo de entidad (Single.map(), Observable.map(), …).

Corutinas de Kotlin + Flujo no necesito tantos operadores, veamos por qué con un ejemplo de los operadores más comunes

mapa()

RxJava:

fun getPerson(id: String): Single
fun observePersons(): Observable

fun getPersonName(id: String): Single 
  return getPerson(id)
     .map  it.firstName 


fun observePersonsNames(): Observable 
  return observePersons()
     .map  it.firstName 

Corutinas de Kotlin + Flujo

suspend fun getPerson(id: String): Person
fun observePersons(): Flow

suspend fun getPersonName(id: String): String? 
  return getPerson(id).firstName


fun observePersonsNames(): Flow 
  return observePersons()
     .map  it.firstName 

No necesita un operador para el caso “único” y es bastante similar para el Flow caso.

mapa plano()

Digamos que necesita, para cada persona, tomar de una base de datos (o servicio remoto) su seguro

RxJava

fun fetchInsurance(insuranceId: String): Single

fun getPersonInsurance(id: String): Single 
  return getPerson(id)
    .flatMap  person ->
      fetchInsurance(person.insuranceId)
    


fun obseverPersonsInsurances(): Observable 
  return observePersons()
    .flatMap  person ->
      fetchInsurance(person.insuranceId) // this is a Single
          .toObservable() // flatMap expect an Observable
    

Veamos con Kotlin Coroutiens + Flow

suspend fun fetchInsurance(insuranceId: String): Insurance

suspend fun getPersonInsurance(id: String): Insurance 
  val person = getPerson(id)
  return fetchInsurance(person.insuranceId)


fun obseverPersonsInsurances(): Flow 
  return observePersons()
    .map  person ->
      fetchInsurance(person.insuranceId)
    

Como antes, con el caso de la co-rutina simple no necesitamos operadores, simplemente escribimos el código como lo haríamos si no fuera asíncrono, simplemente usando funciones de suspensión.

Y con Flow eso NO es un error tipográfico, no es necesario flatMap operador, podemos usar map. ¡Y la razón es que map lambda es una función de suspensión! ¡Podemos ejecutar código de suspensión en él!

No necesitamos otro operador solo para eso.

Para cosas más complejas, puede usar Flow transform() operador.

¡Todos los operadores de Flow aceptan una función de suspensión!

así que si necesitas filter() pero su filtro necesita realizar una llamada de red, ¡puede hacerlo!

fun observePersonsWithValidInsurance(): Flow 
  return observerPersons()
    .filter  person ->
        val insurance = fetchInsurance(person.insuranceId)
        insurance.isValid()
    

delay (), startWith (), concatWith (), …

En RxJava tiene muchos operadores para aplicar demoras o agregar elementos antes y después:

• demora()
• delaySubscription ()
• startWith (T)
• startWith (observable)
• concatCon (…)

con kotlin Flow puedes simplemente:

grabMyFlow()
  .onStart 
    // delay by 3 seconds before starting
    delay(3000L)
    // just emitting an item first
    emit("First item!")
    emit(cachedItem()) // call another suspending function and emit the result
  
  .onEach  value ->
    // insert a delay of 1 second after a value only on some condition
    if (value.length() > 5) 
      delay(1000L)
    
  
  .onCompletion 
    val endingSequence: Flow = grabEndingSequence()
    emitAll(endingSequence)
  

manejo de errores

RxJava tiene muchos operadores para manejar errores:

• onErrorResumeWith ()
• onErrorReturn ()
• onErrorComplete ()

con Flow no necesita mucho más que el operador catch():

  grabMyFlow()
    .catch  error ->
       // emit something from the flow
       emit("We got an error: $error.message")
       // then if we can recover from this error emit it
       if (error is RecoverableError) 
          // error.recover() here is supposed to return a Flow<> to recover
          emitAll(error.recover())
        else 
          // re-throw the error if we can't recover (aka = don't catch it)
          throw error
       
    

y con la función de suspensión solo puedes usar try catch() .

Operadores de flujo fáciles de escribir

Debido a las corrutinas que alimentan Flow bajo el capó, es mucho más fácil escribir operadores. Si alguna vez revisó un operador de RxJava, vería lo difícil que es y cuántas cosas necesita aprender.

Escribir operadores Kotlin Flow es más fácil, puedes hacerte una idea con solo mirar el código fuente de los operadores que ya forman parte de Flow aquí. La razón es que las corrutinas facilitan la escritura de código asíncrono y los operadores simplemente se sienten más naturales de usar.

Como beneficio adicional, los operadores de flujo son todas funciones de extensión de kotlin, lo que significa que usted o las bibliotecas pueden agregar operadores fácilmente y no se sentirán extraños de usar (ej. observable.lift() o observable.compose()).

El hilo ascendente no tiene fugas descendentes

¿Qué significa esto incluso?

Tomemos este ejemplo de RxJava:

urlsToCall()
  .switchMap  url ->
    if (url.scheme == "local") 
       val data = grabFromMemory(url.path)
       Flowable.just(data)
     else 
       performNetworkCall(url)
        .subscribeOn(Subscribers.io())
        .toObservable()
    
  
  .subscribe 
    // in which thread is this call executed?
  

Entonces, ¿dónde está la devolución de llamada? subscribe ¿ejecutado?

La respuesta es:

depende …

si proviene de la red, está en un hilo IO; si proviene de la otra rama, no está definido, depende de qué hilo se use para enviar la URL.

Este es el concepto de “subprocesos ascendentes con fugas descendentes”.

Con Flow y Coroutines, este no es el caso, a menos que requiera explícitamente este comportamiento (usando Dispatchers.Unconfined).

suspend fun myFunction() 
  // execute this coroutine body in the main thread
  withContext(Dispatchers.Main) 
    urlsToCall()
      .conflate() // to achieve the effect of switchMap
      .transform  url ->
        if (url.scheme == "local") 
           val data = grabFromMemory(url.path)
           emit(data)
         else 
           withContext(Dispatchers.IO) 
             performNetworkCall(url)
           
        
      
      .collect 
        // this will always execute in the main thread
        // because this is where we collect,
        // inside withContext(Dispatchers.Main)
      
  

El código de las corrutinas se ejecutará en el contexto en el que se han ejecutado. Y solo la parte con la llamada de red se ejecutará en el subproceso IO, mientras que todo lo demás que vemos aquí se ejecutará en el subproceso principal.

Bueno, en realidad, no sabemos dónde está el código dentro grabFromMemory() se ejecutará, si es una función de suspensión, solo sabemos que se llamará dentro del hilo principal, pero dentro de esa función de suspensión podríamos tener otro Dispatcher en uso, pero cuándo volverá con el resultado val data esto estará nuevamente en el hilo principal.

Lo que significa que, mirando un fragmento de código, es más fácil saber en qué subproceso se ejecutará, si ve un Despachador explícito = es ese despachador, si no lo ve: en cualquier despachador de subproceso la llamada de suspensión que está mirando está siendo llamado.

Simultaneidad estructurada

Este no es un concepto inventado por kotlin, pero es algo que adoptaron más que cualquier otro idioma que conozca.

Si lo que te explico aquí no te alcanza, lee este artículo o mira este video.

¿Así que qué es lo?

Con RxJava te suscribes a observables y te dan una Disposable objeto.

Debe encargarse de desecharlo cuando ya no sea necesario. Entonces, lo que suele hacer es mantener una referencia a él (o ponerlo en un CompositeDisposable) para llamar más tarde dispose() en él cuando ya no sea necesario. Si no lo hace, el linter le dará una advertencia.

RxJava es algo más agradable que un hilo tradicional. Cuando creas un nuevo hilo y ejecutas algo en él, es un “dispara y olvídate”, ni siquiera tienes una forma de cancelarlo: Thread.stop() está en desuso, es dañino y la implementación reciente en realidad no hace nada. Thread.interrupt() hace que tu hilo falle, etc. Cualquier excepción se pierde … te haces una idea.

Con las corrutinas y el flujo de Kotlin, invierten el concepto de “desechables”. NO PUEDE crear una corrutina sin un CoroutineContext.

Este contexto define el scope de su corrutina. Cada corrutina secundaria generada dentro de esa compartirá el mismo alcance.

Si se suscribe a un flujo, debe estar dentro de una corrutina o proporcionar un alcance también.

Todavía puede mantener la referencia de las corrutinas que comienza (Job) y cancelarlos. Esto cancelará todos los elementos secundarios de esa corrutina automáticamente.

Si eres un desarrollador de Android, te dan estos alcances automáticamente. Ejemplo: viewModelScope y puede lanzar corrutinas dentro de un modelo de vista con ese alcance sabiendo que se cancelarán automáticamente cuando se borre el modelo de vista.

viewModelScope.launch 
  // my coroutine here

Algún alcance terminará si algún niño falla, algún otro alcance permitirá que cada niño deje su propio ciclo de vida sin detener a otros niños si uno falla (SupervisedJob).

¿Por qué es esto algo bueno?

Déjame intentar explicarlo como Roman Elizarov hizo.

Algún lenguaje de programación antiguo tenía este concepto de goto que básicamente le permite saltar de una línea de código a otra a voluntad.

Muy poderoso, pero si se abusa, podría terminar con un código muy difícil de entender, difícil de depurar y razonar.

Entonces, los nuevos lenguajes de programación eventualmente lo eliminaron por completo del lenguaje.

Cuando usas if o while o when es mucho más fácil razonar en el código: no importa lo que suceda dentro de esos bloques, eventualmente saldrás de ellos, es un “contexto”, no tienes saltos extraños dentro y fuera.

Lanzar un hilo o suscribirse a un observable RxJava es similar al goto: está ejecutando un código que continuará hasta que se detenga “otro lugar”.

Con las corrutinas, al exigirle que proporcione un contexto / alcance, sabe que cuando su alcance esté sobre todo lo que hay dentro, las corrutinas se completarán cuando se complete su contexto, no importa si tiene una sola corrutina o 10 mil.

Aún puede “ir a” con corrutinas usando GlobalScope, que no deberías por la misma razón que no deberías usar goto en los idiomas que lo proporciona.

¿Algún inconveniente?

Flow todavía está en desarrollo y algunas funciones disponibles en RxJava en este momento todavía no están disponibles en Kotlin Coroutines Flow.

La gran falta, ahora mismo, es share() operadores y sus amigospublish(), replay() etc …)

En realidad, se encuentran en un estado avanzado de desarrollo y se espera que se publiquen pronto (poco después del lanzamiento de kotlin 1.4.0), puede ver el diseño de la API aquí:

Puedes añadir valor a nuestro contenido añadiendo tu veteranía en las interpretaciones.

---

---

Utiliza Nuestro Buscador