Concurrencia
## Objetivos de Aprendizaje
- Comprender conceptos basicos de concurrencia
- Trabajar con Futures
- Aprender introduccion a Akka
- Manejar resultados asincronos
- Comprender conceptos basicos de seguridad de hilos
## Basicos de Concurrencia
### Problemas con Hilos Tradicionales
```scala
// Problema: estado mutable compartido
var counter = 0
val t1 = new Thread(() => {
for (i <- 1 to 10000) counter += 1
})
val t2 = new Thread(() => {
for (i <- 1 to 10000) counter += 1
})
t1.start(); t2.start()
t1.join(); t2.join()
println(counter) // No deterministico! (tipicamente menor que 20000)
```
### Enfoque de Concurrencia de Scala
- **Futures** para operaciones asincronas
- **Promises** para completar futures
- **Akka** para concurrencia basada en actores
- **scala.concurrent** para construcciones asincronas basicas
## Futures
### Creando Futures
```scala
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Success, Failure}
val future = Future {
Thread.sleep(1000)
42
}
```
### Callbacks de Future
```scala
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
val future = Future {
42
}
// onComplete
future.onComplete {
case Success(value) => println(s"Got: $value")
case Failure(ex) => println(s"Error: ${ex.getMessage}")
}
// onSuccess (deprecado pero aun funciona)
future.foreach(value => println(s"Got: $value"))
// onFailure
future.failed.foreach(ex => println(s"Failed: ${ex.getMessage}"))
```
### Bloqueo
```scala
import scala.concurrent.Await
import scala.concurrent.duration._
val future = Future {
Thread.sleep(1000)
42
}
// Bloqueo (evitar cuando sea posible)
val result = Await.result(future, 5.seconds) // 42
```
### map y flatMap en Future
```scala
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
val future = Future { 21 }
// map
val doubled = future.map(_ * 2) // Future(42)
// flatMap
def fetchUser(id: Int): Future[String] = Future(s"User_$id")
val composed = Future(1).flatMap(id => fetchUser(id)) // Future("User_1")
```
### For Comprehensions con Future
```scala
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
def fetchUser(id: Int): Future[String] = Future(s"User_$id")
def fetchEmail(user: String): Future[String] = Future(s"$user@example.com")
val result = for {
user <- fetchUser(1)
email <- fetchEmail(user)
} yield s"$user -> $email"
// Result: Future("User_1 -> User_1@example.com")
```
## Promises
Las promises proporcionan una forma de completar un Future:
```scala
import scala.concurrent.Promise
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
val promise = Promise[Int]()
val future = promise.future
// Completar exitosamente
promise.success(42)
// Completar con falla
promise.failure(new Exception("Oops"))
// Usando tryComplete
promise.tryComplete(Success(42))
promise.tryComplete(Failure(new Exception("Oops")))
```
### Casos de Uso de Promise
```scala
import scala.concurrent.Promise
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
def asyncOperation(callback: Int => Unit): Unit = {
Thread.sleep(1000)
callback(42)
}
def toFuture[A](block: (A => Unit) => Unit): Future[A] = {
val promise = Promise[A]()
block { result =>
promise.success(result)
}
promise.future
}
val future = toFuture[Int](asyncOperation)
```
## Seguridad de Hilos
### Datos Inmutables
```scala
// Case class inmutable - inherentemente thread-safe
case class Person(name: String, age: Int)
// Multiples hilos pueden compartir de forma segura
val persons = List(Person("Alice", 30), Person("Bob", 25))
```
### Operaciones Atomicas
```scala
import scala.concurrent.atomic._
val atomicCounter = AtomicInteger(0)
// Incremento atomico
atomicCounter.incrementAndGet()
// Actualizacion atomica
atomicCounter.updateAndGet(_ + 10)
```
### Synchronized
```scala
class Counter {
private var count = 0
def increment(): Unit = synchronized {
count += 1
}
def get: Int = synchronized {
count
}
}
```
## Introduccion a Akka
Akka es un toolkit para construir aplicaciones concurrentes y distribuidas:
```scala
// build.sbt
// libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.6.x"
```
### Actores
Los actores son objetos que se comunican via mensajes:
```scala
import akka.actor._
// Definir mensajes
case class Greet(name: String)
case class Greeting(message: String)
// Definir actor
class GreeterActor extends Actor {
def receive: Receive = {
case Greet(name) =>
sender() ! Greeting(s"Hello, $name!")
}
}
// Crear y usar
val system = ActorSystem("HelloSystem")
val greeter = system.actorOf(Props[GreeterActor](), "greeter")
greeter ! Greet("World") // Enviar mensaje
// Recibir respuesta
implicit val timeout = Timeout(5.seconds)
val future = greeter ? Greet("World")
val result = Await.result(future, timeout.duration).asInstanceOf[Greeting]
```
### Jerarquia de Actores
```scala
import akka.actor._
class ParentActor extends Actor {
val child = context.actorOf(Props[ChildActor](), "child")
def receive: Receive = {
case msg => child forward msg
}
}
```
## Mejores Practicas
1. **Preferir estructuras de datos inmutables**
2. **Usar Future para operaciones asincronas**
3. **Evitar bloqueo cuando sea posible**
4. **Usar Promises con cuidado**
5. **Considerar Akka para concurrencia compleja**
6. **Usar timeouts para evitar espera indefinida**
7. **Manejar fallas explicitamente**
## Patrones Comunes
### Patron Retry
```scala
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
def retry[T](maxAttempts: Int)(block: => Future[T]): Future[T] = {
block.recoverWith {
case _ if maxAttempts > 1 =>
retry(maxAttempts - 1)(block)
}
}
val result = retry(3) {
Future {
if (math.random() < 0.5) throw new Exception("Random failure")
42
}
}
```
### Patron Timeout
```scala
import scala.concurrent.{Future, TimeoutException}
import scala.concurrent.duration._
import scala.util.{Success, Failure}
def withTimeout[T](future: Future[T], timeout: FiniteDuration): Future[T] = {
Future.firstCompletedOf(List(future, Future {
throw new TimeoutException
}(global)))
}
```
## Resumen
- Scala proporciona Futures para programacion asincrona
- Usar `onComplete`, `foreach` para manejar resultados de Future
- `map` y `flatMap` componenen operaciones de Future
- Las for comprehensions proporcionan codigo asincrono limpio
- Las Promises completan Futures manualmente
- Los datos inmutables son inherentemente thread-safe
- Usar operaciones atomicas o synchronized para estado mutable compartido
- Akka proporciona concurrencia basada en actores para sistemas complejos
- Siempre usar timeouts para prevenir espera indefinida
Comments
Comments powered by Giscus
To enable comments, add your Giscus embed code here.
Learn more about Giscus →