IO asynchrone en Scala avec futures

Question

Disons que je reçois une liste (potentiellement importante) d'images à télécharger à partir de certaines URL. J'utilise Scala, donc ce que je ferais est :

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 _))

Je suis un peu nouveau dans le domaine de Scala, donc cela ressemble encore un peu à de la magie pour moi :

• Est-ce la bonne façon de procéder ? Y a-t-il d'autres solutions si ce n'est pas le cas ?
• Si j'ai 100 images à télécharger, cela créera-t-il 100 fils de discussion à la fois, ou utilisera-t-il un pool de fils de discussion ?
• Est-ce que la dernière instruction ( display _ ) soit exécuté sur le fil principal, et si non, comment puis-je m'assurer qu'il le soit ?

Merci pour vos conseils !

Answer 1

Utilisez les Futures dans Scala 2.10. Il s'agissait d'un travail conjoint entre l'équipe Scala, l'équipe Akka et Twitter pour parvenir à une API et une implémentation futures plus standardisées pour une utilisation dans tous les frameworks. Nous venons de publier un guide à : http://docs.scala-lang.org/overviews/core/futures.html

En plus d'être complètement non bloquants (par défaut, bien que nous fournissions la possibilité d'effectuer des opérations bloquantes gérées) et composables, les futures de Scala 2.10 sont livrés avec un pool de threads implicite pour exécuter vos tâches, ainsi que quelques utilitaires pour gérer les délais.

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)

Ci-dessus, nous importons d'abord un certain nombre de choses :

• future : API pour créer un avenir.
• blocking : API pour le blocage géré.
• Future : Objet compagnon du futur qui contient un certain nombre de méthodes utiles pour collections d'avenirs.
• Await Objet singleton utilisé pour bloquer un futur (en transférant son résultat au thread actuel).
• ExecutionContext.Implicits.global le pool de threads global par défaut, un pool ForkJoin.
• duration._ : des utilitaires pour gérer les durées des interruptions de service.

imagesFuts reste en grande partie la même que ce que vous avez fait à l'origine - la seule différence ici est que nous utilisons le blocage géré -. blocking . Il informe le pool de threads que le bloc de code que vous lui transmettez contient des opérations longues ou bloquantes. Cela permet au pool de créer temporairement de nouveaux travailleurs pour s'assurer que tous les travailleurs ne sont jamais bloqués. Ceci est fait pour éviter la famine (blocage du pool de threads) dans les applications bloquantes. Notez que le pool de threads sait également quand le code d'un bloc bloquant géré est terminé - il supprimera donc le thread ouvrier disponible à ce moment-là, ce qui signifie que le pool sera ramené à sa taille prévue.

(Si vous voulez absolument empêcher la création de threads supplémentaires, vous devez utiliser une bibliothèque AsyncIO, comme la bibliothèque NIO de Java).

Ensuite, nous utilisons les méthodes de collecte de l'objet compagnon Future pour convertir les données de la base de données en données de la base de données. imagesFuts de List[Future[...]] à un Future[List[...]] .

El Await est la façon dont nous pouvons nous assurer que display est exécuté sur le thread appelant-- Await.result force simplement le thread actuel à attendre que le futur qui lui est transmis soit terminé. (Ceci utilise le blocage géré en interne).

Answer 2

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

1. Utilisez scala.concurrent.Future dans la version 2.10, qui est RC maintenant.
2. qui utilise un ExecutionContext implicite
3. La nouvelle doc Future est explicite sur le fait que onComplete (et foreach) peut être évalué immédiatement si la valeur est disponible. Les anciens acteurs Future font la même chose. En fonction de vos exigences en matière d'affichage, vous pouvez fournir un ExecutionContext approprié (par exemple, un exécuteur à thread unique). Si vous voulez simplement que le thread principal attende la fin du chargement, traverse vous donne un futur sur lequel attendre.

Answer 3

1. Oui, ça me semble bien, mais vous pourriez vouloir enquêter sur un système plus puissant. twitter-util o Akka Les API futures (Scala 2.10 disposera d'une nouvelle bibliothèque Future dans ce style).

2. Il utilise un pool de threads.

3. Non, ça n'arrivera pas. Vous devez utiliser le mécanisme standard de votre boîte à outils GUI pour cela ( SwingUtilities.invokeLater pour Swing ou Display.asyncExec pour SWT). Par exemple

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