Think of a scenario where you are standing in front of a long queue of your cafeteria to order your favorite food. Some people might get so frustrated that they leave the queue without even ordering. Thinking of these types of situations cafeteria management decided to introduce a token system. You can simply sit and chit-chat with your friends while waiting for your token number. A similar principle is adopted in the Akka Actor Model.
Akka actors use java threads internally. More than a few thousand threads running simultaneously produces overhead on most desktop machines. Akka Actors simplifies this problem by providing awake-on-demand actors who do not occupy a thread unless they have some work to do. Just like cafeteria scenario, unless you do not get a notification of your token number you can continue to perform your work.
Akka provides open-source libraries for designing scalable and resilient systems that are capable to focus on business needs that require reliable, fault tolerance and high-performance behavior.
- Fault-Tolerant – This can be clearly understood with the motto of Akka being “Let it fail”. Akka provides *Actor System* having supervisor hierarchies along with application logic can help in dealing with failures.
- Scaling – The majority of innovation in Akka lies in removing the bits that didn’t cluster well and providing support for deep clustering.
- Ease of deployment – Recently launched ConductR1.0, commercial technology designed to let DevOps team manage the deployment of Reactive, Distributed applications using Java or Scala that are built along with Play, Akka and Slick.
Implementing Java Threads and Akka Actors
Let us start with an assignment of reading a file of size 51.5 MB with 963366 words and counting the number of words with it using different versions of concurrency.
Plan 1: Make a multithreaded version of threads in java
I started with a simple multi-threaded java program which consists of 50 threads. All these threads are simply responsible for calculating the number of words in a file having 1913714 words and size 100 MB.
| import java.io.BufferedReader; | |
| import java.io.File; | |
| import java.io.FileReader; | |
| public class Count implements Runnable { | |
| private static long processedTime = 0; | |
| File file; | |
| public Count(File fileName) { | |
| file = fileName; | |
| } | |
| public static void main(String[] args) throws InterruptedException { | |
| File file = new File("/home/workspace/enwik8"); | |
| Runnable runnable = new Count(file); | |
| long startTime = System.currentTimeMillis(); | |
| Thread thread; | |
| for (int i = 0; i < 50; i++) { | |
| thread =new Thread(runnable); | |
| thread.start(); | |
| thread.join(); | |
| } | |
| long elapsedTime = System.currentTimeMillis() – startTime; | |
| System.out.println("Threads processing time " + elapsedTime + "milliseconds"); | |
| } | |
| public void run() { | |
| int i = 0; | |
| int wordCount = 0; | |
| try { | |
| FileReader reader = new FileReader(file); | |
| BufferedReader buffer = new BufferedReader(reader); | |
| String line; | |
| while ((line = buffer.readLine()) != null) { | |
| String[] wordList = line.split(","); | |
| wordCount += wordList.length; | |
| } | |
| System.out.println("Number of words in " + Thread.currentThread().getName() + "-" + wordCount); | |
| reader.close(); | |
| } | |
| catch (Exception e) { | |
| System.out.println(e); | |
| } | |
| } | |
| } |
Total thread processing time for 50 threads appear to be 29174 milliseconds ms(without synchronised). It appears to be 30317 milliseconds ms(with synchronised). This happens, because locks due to synchronised introduce a new menace: Deadlock. With multiple locks in place, performance suffers and processing time increases.
In this scenario, you also need to worry about how the interleaved execution of multiple threads fulfills the invariant which is processing the complete file at once takes place.
Plan 2: Make an actor using scala
Implementing the same scenario using the actor system in scala:-
| import akka.Done | |
| import akka.actor.Actor.Receive | |
| import akka.actor.{Actor, ActorContext, ActorSystem, Props} | |
| import akka.pattern.ask | |
| import akka.routing.RoundRobinPool | |
| import akka.util.Timeout | |
| import scala.concurrent.Future | |
| import scala.concurrent.duration._ | |
| import scala.io.Source | |
| import scala.language.postfixOps | |
| class Counter extends Actor { | |
| override def receive: Receive = TimingReceive { | |
| case file: String => | |
| val content = Source.fromFile(file) | |
| .getLines().toList | |
| .map(row => row.split(",").toList) | |
| .size | |
| println("Number of words in file " + file + "-" + content) | |
| sender() ! Done | |
| case _ => None | |
| } | |
| } | |
| class TimingReceive(r: Receive, totalTime: Long)(implicit ctx: ActorContext) extends Receive { | |
| def isDefinedAt(o: Any): Boolean = { | |
| r.isDefinedAt(o) | |
| } | |
| def apply(o: Any): Unit = { | |
| val startTime = System.nanoTime() | |
| r(o) | |
| val newTotal = totalTime + (System.nanoTime() – startTime) | |
| println("Total time so far: " + totalTime + " milliseconds") | |
| ctx.become(new TimingReceive(r, newTotal)) | |
| } | |
| } | |
| object WordCountUsingAkka { | |
| implicit val timeout: Timeout = Timeout(50 seconds) | |
| def main(args: Array[String]) { | |
| val system = ActorSystem("Counting") | |
| val props = Props[Counter].withRouter(RoundRobinPool(2)) | |
| val actor = system.actorOf(props, "Ping") | |
| val result: List[Future[Done]] = List.fill[String](10)("/home/workspace/Desktop/enwik8").map { file => | |
| (actor ? file).mapTo[Done] | |
| } | |
| } | |
| } | |
| object TimingReceive { | |
| def apply(r: Receive)(implicit ctx: ActorContext): Receive = new TimingReceive(r, 0) | |
| } |
Include below dependency in build.sbt:-
libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.5.25"
Upon execution of this code, it uses 9990.50850 ms to count the number of words in a file. It uses a round-robin pool of 2 actors and routes 10 files having 1128024 words to all the actors. Increasing actors will require more heap size but reduces processing time.
Summary
To Sum up the actor model simulated above clearly depicts the following pointers:-
- Encapsulation without resorting to locks.
- You do not need to worry about the executing mechanism of actors.
- It provides a model of communicating entities, that responds to messages, changes states and driving the whole application forward.
- Usage of message passing avoids locking and blocking. – As we saw in the above example, we are simply sending messages to Counter and it does not transfer thread of execution to destination. An actor can send a message and continue without blocking.
- An important difference noted here is the first plan was calling a run method for every thread. In the second plan, the message is being passed to the actor. The actor delegates work to another actor by sending a message.
- For a more detailed understanding of Akka actor model refer to the link below
https://doc.akka.io/docs/akka/2.5.3/scala/guide/actors-intro.html
Thanks for reading. Stay Tuned for upcoming blogs on Akka streams. 🙂
