Concurrency
## Learning Objectives
- Understand basic concurrency concepts
- Work with Futures
- Learn about Akka introduction
- Handle async results
- Understand thread safety basics
## Concurrency Basics
### Problems with Traditional Threads
```scala
// Problem: Mutable shared state
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) // Non-deterministic! (typically less than 20000)
```
### Scala's Concurrency Approach
- **Futures** for async operations
- **Promises** for completing futures
- **Akka** for actor-based concurrency
- **scala.concurrent** for basic async constructs
## Futures
### Creating Futures
```scala
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Success, Failure}
val future = Future {
Thread.sleep(1000)
42
}
```
### Future Callbacks
```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 (deprecated but still works)
future.foreach(value => println(s"Got: $value"))
// onFailure
future.failed.foreach(ex => println(s"Failed: ${ex.getMessage}"))
```
### Blocking
```scala
import scala.concurrent.Await
import scala.concurrent.duration._
val future = Future {
Thread.sleep(1000)
42
}
// Blocking (avoid when possible)
val result = Await.result(future, 5.seconds) // 42
```
### map and flatMap on 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 with 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
Promises provide a way to complete a Future:
```scala
import scala.concurrent.Promise
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
val promise = Promise[Int]()
val future = promise.future
// Complete successfully
promise.success(42)
// Complete with failure
promise.failure(new Exception("Oops"))
// Using tryComplete
promise.tryComplete(Success(42))
promise.tryComplete(Failure(new Exception("Oops")))
```
### Promise Use Cases
```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)
```
## Thread Safety
### Immutable Data
```scala
// Immutable case class - inherently thread-safe
case class Person(name: String, age: Int)
// Multiple threads can safely share
val persons = List(Person("Alice", 30), Person("Bob", 25))
```
### Atomic Operations
```scala
import scala.concurrent.atomic._
val atomicCounter = AtomicInteger(0)
// Atomic increment
atomicCounter.incrementAndGet()
// Atomic update
atomicCounter.updateAndGet(_ + 10)
```
### Synchronized
```scala
class Counter {
private var count = 0
def increment(): Unit = synchronized {
count += 1
}
def get: Int = synchronized {
count
}
}
```
## Akka Introduction
Akka is a toolkit for building concurrent, distributed applications:
```scala
// build.sbt
// libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.6.x"
```
### Actors
Actors are objects that communicate via messages:
```scala
import akka.actor._
// Define messages
case class Greet(name: String)
case class Greeting(message: String)
// Define actor
class GreeterActor extends Actor {
def receive: Receive = {
case Greet(name) =>
sender() ! Greeting(s"Hello, $name!")
}
}
// Create and use
val system = ActorSystem("HelloSystem")
val greeter = system.actorOf(Props[GreeterActor](), "greeter")
greeter ! Greet("World") // Send message
// Receive response
implicit val timeout = Timeout(5.seconds)
val future = greeter ? Greet("World")
val result = Await.result(future, timeout.duration).asInstanceOf[Greeting]
```
### Actor Hierarchy
```scala
import akka.actor._
class ParentActor extends Actor {
val child = context.actorOf(Props[ChildActor](), "child")
def receive: Receive = {
case msg => child forward msg
}
}
```
## Best Practices
1. **Prefer immutable data structures**
2. **Use Future for async operations**
3. **Avoid blocking when possible**
4. **Use Promises carefully**
5. **Consider Akka for complex concurrency**
6. **Use timeouts to avoid indefinite waiting**
7. **Handle failures explicitly**
## Common Patterns
### Retry Pattern
```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
}
}
```
### Timeout Pattern
```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)))
}
```
## Summary
- Scala provides Futures for async programming
- Use `onComplete`, `foreach` to handle Future results
- `map` and `flatMap` compose Future operations
- for comprehensions provide clean async code
- Promises manually complete Futures
- Immutable data is inherently thread-safe
- Use atomic operations or synchronized for mutable shared state
- Akka provides actor-based concurrency for complex systems
- Always use timeouts to prevent indefinite waiting
Comments
Comments powered by Giscus
To enable comments, add your Giscus embed code here.
Learn more about Giscus →