multithreading - IO asincrónico en Scala con futuros

future

Question

Digamos que obtengo una lista (potencialmente grande) de imágenes para descargar desde algunas URL. Estoy usando Scala, entonces lo que haría es:

import scala.actors.Futures._

// Retrieve URLs from somewhere
val urls: List[String] = ...

// Download image (blocking operation)
val fimages: List[Future[...]] = urls.map (url => future { download url })

// Do something (display) when complete
fimages.foreach (_.foreach (display _))

Soy un poco nuevo para Scala, así que esto todavía se parece un poco a la magia para mí:

• ¿Esta es la manera correcta de hacerlo? Cualquier alternativa si no es?
• Si tengo 100 imágenes para descargar, ¿creará 100 hilos a la vez, o usará un grupo de hilos?
• ¿Se ejecutará la última instrucción ( display _ ) en el hilo principal, y si no es así, cómo puedo asegurarme de que sea?

¡Gracias por su consejo!

Answer 1

1. Sí, me parece bien, pero es posible que desee investigar twitter-util o Akka Future más potentes (Scala 2.10 tendrá una nueva biblioteca de Future en este estilo).

2. Utiliza un grupo de hilos.

3. No, no lo hará. Debe usar el mecanismo estándar de su kit de herramientas GUI para esto ( SwingUtilities.invokeLater for Swing o Display.asyncExec for SWT). P.ej

   fimages.foreach (_.foreach(im => SwingUtilities.invokeLater(new Runnable { display im })))

Answer 2

Use futuros en Scala 2.10. Fueron un trabajo conjunto entre el equipo de Scala, el equipo de Akka y Twitter para alcanzar una API e implementación futuras más estandarizadas para su uso en todos los marcos. Acabamos de publicar una guía en: http://docs.scala-lang.org/overviews/core/futures.html

Más allá de ser completamente no bloqueante (de manera predeterminada, aunque ofrecemos la capacidad de realizar operaciones de bloqueo administradas) y composable, los futuros de Scala 2.10 vienen con un grupo de hilos implícitos para ejecutar sus tareas, así como algunas utilidades para administrar los tiempos de espera.

import scala.concurrent.{future, blocking, Future, Await, ExecutionContext.Implicits.global}
import scala.concurrent.duration._

// Retrieve URLs from somewhere
val urls: List[String] = ...

// Download image (blocking operation)
val imagesFuts: List[Future[...]] = urls.map {
  url => future { blocking { download url } }
}

// Do something (display) when complete
val futImages: Future[List[...]] = Future.sequence(imagesFuts)
Await.result(futImages, 10 seconds).foreach(display)

Arriba, primero importamos una cantidad de cosas:

• future : API para crear un futuro.
• blocking : API para bloqueo administrado.
• Future : objeto compañero futuro que contiene varios métodos útiles para colecciones de futuros.
• Await : objeto singleton utilizado para bloquear en un futuro (transfiriendo su resultado al hilo actual).
• ExecutionContext.Implicits.global : el grupo de subprocesos global predeterminado, un grupo ForkJoin.
• duration._ : utilidades para gestionar las duraciones de tiempos muertos.

imagesFuts sigue siendo básicamente el mismo que originalmente: la única diferencia aquí es que utilizamos el bloqueo de blocking gestionado. Notifica al grupo de subprocesos que el bloque de código que le pasa contiene operaciones de larga ejecución o de bloqueo. Esto permite que el grupo genere temporalmente nuevos trabajadores para asegurarse de que nunca ocurra que todos los trabajadores estén bloqueados. Esto se hace para evitar la inanición (bloqueando el grupo de subprocesos) en aplicaciones de bloqueo. Tenga en cuenta que el grupo de subprocesos también sabe cuándo se completa el código en un bloque de bloqueo administrado, por lo que eliminará el subproceso de trabajador de repuesto en ese punto, lo que significa que el grupo se reducirá a su tamaño esperado.

(Si quiere evitar absolutamente que se creen hilos adicionales, entonces debe usar una biblioteca AsyncIO, como la biblioteca NIO de Java).

Luego usamos los métodos de recopilación del objeto acompañante Futuro para convertir imagesFuts de List[Future[...]] a Future[List[...]] .

El objeto Await es cómo podemos asegurarnos de que la display se ejecuta en el hilo de llamada. Await.result simplemente fuerza a que el hilo actual espere hasta que se complete el futuro en el que se pasó. (Esto usa bloqueo administrado internamente).

Answer 3

val all = Future.traverse(urls){ url =>
  val f = future(download url) /*(downloadContext)*/
  f.onComplete(display)(displayContext)
  f
}
Await.result(all, ...)

1. Use scala.concurrent.Future en 2.10, que ahora es RC.
2. que usa un ExecutionContext implícito
3. El nuevo documento de Future es explícito de que onComplete (y foreach) puede evaluar inmediatamente si el valor está disponible. Los viejos actores Future hacen lo mismo. Dependiendo de cuáles sean sus requisitos para la visualización, puede suministrar un ExecutionContext adecuado (por ejemplo, un ejecutor de hilos único). Si solo quiere que el hilo principal espere a que se complete la carga, poligonal le ofrece un futuro en el que esperar.