Akka.io Quickstart With Java – Check My 1 Cool Secret!

Very often you can meet the requirement to create an application that will collect data (Akka.io Quickstart with Java") from an external system through its API. It may turn out that the bottleneck of such an application will be the time of its implementation.

Introduction

Multithread Processing

Multithread processing is a technique used to improve the performance and efficiency of computer programs by allowing them to make use of multiple CPU cores or processors. It involves dividing a program into multiple threads, each of which can be executed concurrently by the CPU.

Multithread processing can be used to improve the performance of programs that perform a lot of computations or that are waiting on external events such as I/O operations. By using multiple threads, the program can perform different tasks simultaneously, taking advantage of idle CPU cores and reducing the overall execution time.

Multithread processing is supported by many programming" languages and operating systems, and it is often used in applications such as web servers, database systems, and scientific simulations that need to process large amounts of data or handle many concurrent requests.

Advantages Of Multithread Processing

There are several advantages to using multithread processing in a computer program:

1. Improved performance: By allowing a program to make use of multiple CPU cores or processors, multithread processing can significantly improve the performance of the program, especially for tasks that are computationally intensive or involve waiting on external events.
2. Increased efficiency: Multithread processing can improve the efficiency of a program by allowing it to take advantage of idle CPU resources and reduce the overall execution time.
3. Better responsiveness: Multithread processing can improve the responsiveness of a program by allowing it to perform background tasks concurrently with the main thread, without blocking or delaying the main thread.
4. Enhanced scalability: Multithread processing can make a program more scalable by allowing it to take advantage of additional CPU resources as they become available, making it easier to handle increases in workload.
5. Simplified programming: In some cases, multithread processing can make it easier to write and maintain programs by allowing different parts of the program to be implemented and tested independently.

Multithread Processing

In Java", multithread processing is supported through the use of threads. A thread is a separate execution path that can be run concurrently with other threads in a program. Java" includes a built-in Thread class and a Thread API that provide a set of methods for creating and managing threads in a program.

To create a new thread in Java", you can either extend the Thread class and override the run() method, or create a new Thread object and pass a Runnable object to its constructor. The Runnable object defines a run() method that contains the code to be executed by the thread.

Once a thread has been created, it can be started by calling its start() method. This will cause the thread to begin execution at the start of the run() method. The thread will continue to run until the run() method completes, or until it is interrupted or terminated by another thread.

Java" also provides a number of methods and synchronization tools for controlling and coordinating the execution of threads, such as wait(), notify(), and synchronized blocks. These can be used to ensure that threads access shared resources safely and correctly.

Multithread processing can be useful for improving the performance and responsiveness of Java" programs, especially for tasks that are computationally intensive or involve waiting on external events. It is commonly used in applications such as web servers, database systems, and scientific simulations that need to process large amounts of data or handle many concurrent requests.

Frameworks

Framework is a toolkit and runtime for building distributed and concurrent applications in Java and Scala". It is designed to simplify the development of scalable and fault-tolerant systems by providing a set of abstractions and patterns for building distributed systems.

Akka IO is a module of framework that provides a set of tools and libraries for building networked applications in Java". It includes a set of actor-based abstractions for working with network sockets, streams, and protocols, as well as a set of utilities for parsing and serializing data to and from various formats.

Using Framework, developers can build applications that can communicate over a network using various protocols, such as TCP, UDP, and HTTP. It can be used to build client and server applications, as well as peer-to-peer systems.

Framework is built on top of the Java" NIO library, which provides a set of non-blocking I/O operations for working with network sockets and streams. It is designed to be lightweight and efficient, and it can be used to build high-performance networked applications in Java".

Dive Into Topic

Let’s assume you need to refresh customer data every 15 minutes. The API we are dealing with does not allow us to send a request that will return information about all clients or groups of clients. We can only send a request where we can provide the customer ID. We have about 10,000 customers.

The average time from sending a query to the response time is 0.4 seconds. As it is easy to count, when we run the application, the time in which we will be able to receive the refresh data will be: 66.66 min, which exceeds the refresh window more than 4 times!. In short, we will not meet the requirements set before us. (Akka.io Quickstart with Java")

One solution can be to run the same application several times, where each instance will receive a different set of customer IDs. But this is not a very elegant solution and causes more problems, because we have to add the logic to divide customers into packages, then run several instances, monitor them, etc. (Akka.io Quickstart with Java")

Akka.io Quickstart With Java

Framework comes to help. Based on the concept of actors, where each actor is run as an independent thread. The implementation of this library is very simple.

We define actors and assign them tasks. We are not interested in concurrency services, thanks to which we can look at this problem from a higher level. Let’s try Akka.io! (Akka.io Quickstart with Java")

Akka Actors Quickstart With Java

1. Concept And Communication In Akka.io

The following figure shows the concept of actors and communication between them, where it is presented as two-way, but of course it can also be unidirectional (it was also implemented in the example included in this post).

The application consists of one actor, who is responsible only for displaying messages in consoles (WorkersAdministrator), which will be sent by actors (workers) in Akka.io. (Akka.io Quickstart with Java")

ActorSystem system = ActorSystem.create("Actor-api-collector");

// Create Actor, which will be responsible only for printing messages from workers
final ActorRef workersAdministrator = system.actorOf(Printer.props(), "workersAdministrator");

// Create set of workers
ActorRef worker1 = system.actorOf(DataCollector.props(workersAdministrator));
ActorRef worker2 = system.actorOf(DataCollector.props(workersAdministrator));
ActorRef worker3 = system.actorOf(DataCollector.props(workersAdministrator));
ActorRef worker4 = system.actorOf(DataCollector.props(workersAdministrator));
ActorRef worker5 = system.actorOf(DataCollector.props(workersAdministrator));

Each Worker will send information to the WorkersAdministrator about the fact that it received the task to retrieve information about the customer about the transferred ID.

workersAdministrator.tell("--> Getting data for ID=[" + id + "] ... ", getSelf());

The download has been completed. In addition, it will tell in the message the time elapsed since the receipt of the task to the end of the download.

workersAdministrator.tell("<-- Data collected! ID=[" + id + "] in time=[" + sleepInt + "] milliseconds.", getSelf());

2. WorkersAdministrtor in Akka Http

Each class that is to be an actor must expand the AbstractActor class. This will force the user" to implement the: createReceive() method. (Akka.io Quickstart with Java")

This method implements a logic that uses matching pattern (it is more known from Scala" than from Java"). We define what class we expect the object to be passed on. In our case, the actors (workers) will transmit a text message, which is why it is of the class String.class, and “message” is our reference to this object.

package com.bigdataetl.akka;

import akka.actor.AbstractActor;
import akka.actor.Props;

public class WorkersAdministrator extends AbstractActor {

    static public Props props() {
        return Props.create(WorkersAdministrator.class, () -> new WorkersAdministrator());
    }

    public WorkersAdministrator() {
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(String.class, message -> {
                    // Just print the incoming messages from the Actors
                    System.out.println(message);
                })
                .build();
    }

}

3. Worker (DataCollector) in Akka.io

Akka.io: Our worker, i.e. an instance of the DataCollector class, must also extend the AbstractActor class. (Akka.io Quickstart with Java")

In the case of Worker, a customer ID is passed, which in our case is a value of Integer type.

To simulate the wait for a server response to a request sent, a private collectData() method was created that returns a random value of milliseconds that will hold the thread (actor) and then return this value so that it can be passed to the administrator. (Akka.io Quickstart with Java")

package com.bigdataetl.akka;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.Props;

import java.util.Random;

public class DataCollector extends AbstractActor {

    private int actorId;
    private ActorRef workersAdministrator;
    private Random rnd = new Random();

    public DataCollector(int actorId, ActorRef workersAdministrator) {
        this.actorId = actorId;
        this.workersAdministrator = workersAdministrator;
    }

    static public Props props(int actorId, ActorRef printerActor) {
        return Props.create(DataCollector.class, () -> new DataCollector(actorId, printerActor));
    }

    // This method is a API mock-up response.
    // We sent the request and we are waiting for response.
    // I used the random class to demonstrate the random time of response
    private int collectData() {
        int sleepTime = rnd.nextInt(300) + 300;
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return sleepTime;
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(Integer.class, clientId -> {
                    // Tell the WorkersAdministrator that this Actor sent the request and is waiting for response.
                    workersAdministrator.tell("--> {Actor " + actorId + "} Sending request for data. ID=[" + clientId + "] ... ", getSelf());

                    // Get the response time and tell the WorkersAdministrator that data was received.
                    int sleepInt = collectData();
                    workersAdministrator.tell("<-- {Actor " + actorId + "} Data collected! ID=[" + clientId + "] in time=[" + sleepInt + "] milliseconds.", getSelf());
                })
                .build();
    }
}

4. Main Class In Akka.io

Akka.io: At the beginning, we register" AktorSystem. Then, the actors will be registered as part of the “system“. (Akka.io Quickstart with Java")

Please note that it’s up to you what the relationship between the actors will be. From the point of view of the library, all actors are equal. It is the developer who sets the tree of relations between them.

Then, in turn, we register" Worker Administrator and 5 actors (DataCollector). (Akka.io Quickstart with Java")

The loop contains a logic that delegates work to individual workers. For simplicity, I assumed that customer IDs would start from 1 to 100. (Akka.io Quickstart with Java")

package com.bigdataetl;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import com.bigdataetl.akka.DataCollector;
import com.bigdataetl.akka.WorkersAdministrator;

public class Main {

    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("Actor-api-collector");

        // Create Actor, which will be responsible only for printing messages from workers
        final ActorRef workersAdministrator = system.actorOf(WorkersAdministrator.props(), "workersAdministrator");

        // Create set of workers
        ActorRef worker1 = system.actorOf(DataCollector.props(1,workersAdministrator));
        ActorRef worker2 = system.actorOf(DataCollector.props(2,workersAdministrator));
        ActorRef worker3 = system.actorOf(DataCollector.props(3,workersAdministrator));
        ActorRef worker4 = system.actorOf(DataCollector.props(4,workersAdministrator));
        ActorRef worker5 = system.actorOf(DataCollector.props(5, workersAdministrator));

        // This loop demonstrate how we can delegate jobs to Actors.
        // We would like to sent the requests to API to get data for each requested ID.
        for (int id = 1; id <= 100; id++) {

            switch (id % 5) {
                case 0:
                    worker1.tell(id, ActorRef.noSender());
                    break;
                case 1:
                    worker2.tell(id, ActorRef.noSender());
                    break;
                case 2:
                    worker3.tell(id, ActorRef.noSender());
                    break;
                case 3:
                    worker4.tell(id, ActorRef.noSender());
                    break;
                case 4:
                    worker5.tell(id, ActorRef.noSender());
                    break;
            }

        }
    }

}

5. Results Of Akka.io Usage

After starting the program, you should get a similar result as below.

As you can see, at the beginning, we only see information that the work was assigned to individual actors, and then in later entries we can see how the information about the response from the server and the data have been retrieved begins to appear. When the actor finishes downloading, he immediately gets another task to do. (Akka.io Quickstart with Java")

--> {Actor 2} Sending request for data. ID=[1] ... 
--> {Actor 3} Sending request for data. ID=[2] ... 
--> {Actor 5} Sending request for data. ID=[4] ... 
--> {Actor 4} Sending request for data. ID=[3] ... 
--> {Actor 1} Sending request for data. ID=[5] ... 
<-- {Actor 5} Data collected! ID=[4] in time=[400] milliseconds.
--> {Actor 5} Sending request for data. ID=[9] ... 
<-- {Actor 4} Data collected! ID=[3] in time=[517] milliseconds.
--> {Actor 4} Sending request for data. ID=[8] ... 
<-- {Actor 2} Data collected! ID=[1] in time=[530] milliseconds.
--> {Actor 2} Sending request for data. ID=[6] ... 
<-- {Actor 3} Data collected! ID=[2] in time=[563] milliseconds.
--> {Actor 3} Sending request for data. ID=[7] ... 
<-- {Actor 1} Data collected! ID=[5] in time=[578] milliseconds.
--> {Actor 1} Sending request for data. ID=[10] ... 
<-- {Actor 5} Data collected! ID=[9] in time=[485] milliseconds.
--> {Actor 5} Sending request for data. ID=[14] ... 
<-- {Actor 3} Data collected! ID=[7] in time=[378] milliseconds.
--> {Actor 3} Sending request for data. ID=[12] ... 
<-- {Actor 1} Data collected! ID=[10] in time=[397] milliseconds.
--> {Actor 1} Sending request for data. ID=[15] ... 
<-- {Actor 4} Data collected! ID=[8] in time=[508] milliseconds.
--> {Actor 4} Sending request for data. ID=[13] ... 
<-- {Actor 2} Data collected! ID=[6] in time=[550] milliseconds.
--> {Actor 2} Sending request for data. ID=[11] ... 
<-- {Actor 1} Data collected! ID=[15] in time=[330] milliseconds.
--> {Actor 1} Sending request for data. ID=[20] ... 
<-- {Actor 5} Data collected! ID=[14] in time=[424] milliseconds.
--> {Actor 5} Sending request for data. ID=[19] ... 
<-- {Actor 3} Data collected! ID=[12] in time=[504] milliseconds.
--> {Actor 3} Sending request for data. ID=[17] ... 
<-- {Actor 4} Data collected! ID=[13] in time=[484] milliseconds.
--> {Actor 4} Sending request for data. ID=[18] ... 
<-- {Actor 2} Data collected! ID=[11] in time=[509] milliseconds.
--> {Actor 2} Sending request for data. ID=[16] ... 
<-- {Actor 1} Data collected! ID=[20] in time=[379] milliseconds.
--> {Actor 1} Sending request for data. ID=[25] ... 
<-- {Actor 5} Data collected! ID=[19] in time=[419] milliseconds.
--> {Actor 5} Sending request for data. ID=[24] ... 
<-- {Actor 4} Data collected! ID=[18] in time=[315] milliseconds.
--> {Actor 4} Sending request for data. ID=[23] ... 
<-- {Actor 3} Data collected! ID=[17] in time=[496] milliseconds.
--> {Actor 3} Sending request for data. ID=[22] ... 
<-- {Actor 2} Data collected! ID=[16] in time=[494] milliseconds.
--> {Actor 2} Sending request for data. ID=[21] ... 
<-- {Actor 1} Data collected! ID=[25] in time=[417] milliseconds.
--> {Actor 1} Sending request for data. ID=[30] ... 
<-- {Actor 5} Data collected! ID=[24] in time=[410] milliseconds.
--> {Actor 5} Sending request for data. ID=[29] ... 
<-- {Actor 4} Data collected! ID=[23] in time=[321] milliseconds.
...

Summary Of Akka.io Usage

Akka library, as you can see, is easy to use, and gives you a thread management system that is not visible to the developer. Using this library can help solve many of the problems that may be unmanageable in nature in a single-threaded mode. (Akka.io Quickstart with Java")

Could You Please Share This Post? 
I appreciate It And Thank YOU! :)
Have A Nice Day!