Why did Flink use Akka instead of RPC?

Problems with the original RPC service:

  • There is no asynchronous invocation of callbacks, which is why Flink’s multiple runtime components require poll state, causing unnecessary delays.
  • Without Exception forwarding, exceptions are simply devoured, which makes for some weird problems that are very difficult to debug at run time
  • The number of threads on the processor is limited, and RPC can only handle a certain number of concurrent requests, forcing you to isolate the thread pool
  • Arguments do not support primitive data types (or boxing types for primitive data types), and everything must have a special serialized class
  • Tricky threading model, where RPC continually creates or terminates threads

Benefits of using Akka’s actor model:

  • Akka solves all of these problems and is transparent
  • The Supervisor model allows you to do failure detection on actors, providing a unified way to detect and process failures (such as lost heartbeats, failed calls…).
  • Akka has tools to persist stateful actors and restart them on other machines if they fail. This mechanism can be very useful and important in master Fail-over scenarios.
  • You can define many call targets (actors), and tasks on TaskManager can call their ExecutionVertex directly on JobManager, rather than calling JobManager to generate a thread to see the execution status.
  • The Actor model is close to running queue models one after another on actors, which makes the concurrent model of a state machine simple and robust

Introduction of Akka

Akka is a framework for developing applications that support concurrency, fault tolerance, and extensibility based on the Actor model.

In the context of the actor model, all active entities are considered to be interdependent actors, and actors communicate with each other by sending asynchronous messages to each other. Each actor has a Mailbox to store the messages it receives, so each actor maintains its own independent state.

Several features of Akka

  1. Simple Concurrency & Distribution easy to Build Parallel and Distributed applications

    Akka adopts asynchronous communication and distributed architecture in its design, and abstracts the upper layer, such as Actors, Futures, STM, etc.

  2. Resilient by Design

    The system has self-healing capability and can be monitored locally or remotely.

  3. High Performance

    It can send 50 million messages per second in a single machine. Small memory footprint, up to 2.5 million actors can be stored in 1GB of memory.

  4. Elastic, no center (Elastic – Decentralized)

    Adaptive balancing, routing, partitioning, configuration

  5. Extensible

    This can be extended using the Akka extension pack.

The core of Akka (Akka-Actor) is very small and can be easily put into your application to provide the asynchronous lockless parallelism you need. It can be used in two different ways:

  • As a library: used in web applications, in web-INF /lib, or in the classpath as a normal Jar package.
  • ** In the form of microkernels: ** You can put applications into a separate kernel. You have a main class to initialize the Actor system.

Akka is easy to use

Build Akka development environment using IDEA+Maven.

  1. Importing Maven dependencies

    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <encoding>UTF-8</encoding>
        <scala.version>2.118.</scala.version>
        <scala.compat.version>2.11</scala.compat.version>
        <akka.version>2.36.</akka.version> </properties> <dependencies> <! Scala -lang</groupId> <artifactId> Scala -library</artifactId> <version>${scala.version}</version> </dependency> <! <dependency> <groupId>com.typesafe.akka</groupId> <artifactId>akka-actor_${scala.compat.version}</artifactId> <version>${akka.version}</version> </dependency> <! <dependency> <groupId>com.typesafe.akka</groupId> <artifactId>akka-remote_${scala.compat.version}</artifactId> <version>${akka.version}</version> </dependency> </dependencies>Copy the code

  2. Java implementation Actor Demo

    / * * *@author li.pan
 * @version 1.0.0
 * @Description TODO
 * @createTime22 December 2020 20:18:00 */
public class JavaPongActor extends AbstractActor {

    /** * The AbstractActor class has a receive method that subclasses must implement or call via a constructor. * *@returnThe type returned is PartialFunction, which comes from the Scala API. In Java * Akka provides us with an abstract constructor class ReceiveBuilder that generates PartialFunction as the return value. * /
    @Override
    public PartialFunction receive(a) {
        return ReceiveBuilder
                // matchEquals and matchAny are used to match messages
                .matchEquals("Ping", s ->
                        sender().tell("Pong", ActorRef.noSender()))
                .match(String.class, s ->
                        System.out.println("It's a string: " + s))
                .matchAny(x ->
                        sender().tell(
                                new Status.Failure(new Exception("unknown message")), self())) .build(); }}Copy the code

    The above Java code shows how to use the corresponding Java API in Akka. Each specific API has the following meanings:

    methods meaning
    Receive The AbstractActor class has a receive method,Its subclasses must implement the method or call it through a constructor.The receive method returns a type of PartialFunction, which comes from the Scala API. In Java, there are no native methods to construct Scala’s partialfunctions, so Akka provides us with an abstract constructor class
    ReceiveBuilder The Build method returns the PartialFunction
    Match Similar to pattern matching in Scala, for matching messages:

    match(class, function): describes how to respond to any unmatched examples of this type.

    match(String.class, s -> {if(s.equals("Ping")) respondToPing(s); })

    match(class, predicate, function): describes how to respond to a certain type of message for which the predicate condition function is true.

    match(String.class, s -> s.equals("Ping"), s -> respondToPing(s))

    matchEquals(object, function): describes how to respond to a message that is equal to the first argument passed in.

    matchEquals("Ping", s -> respondToPing(s))

    matchAny(function): This function matches all unmatched messages. In general, the best practice is to return error messages, or at least log them to aid in error debugging during development.
    Sender Returns a response to the message received, either by an Actor or a request from outside the Actor system.
    Tell The sender() function returns an ActorRef. In the example above, we call sender().tell(). Tell () is the most basic one-way message transfer pattern. The first parameter is the message we want to send to the recipient’s mailbox. The second argument is the sender you want the other Actor to see.
    Ask Send a message to the Actor that returns a Future. When the Actor returns the response, the Future is completed. No messages are returned to the sender’s mailbox.
    Forward Sends the received message to another Actor. All responses sent to sender() are returned to the sender of the original message.
    Pipe Used to return the result of the Future to the sender() or another Actor. If you are using Ask or working with a Future, using Pipe will return the result of the Future correctly.

  3. Scala implements Actor Demo

    class ScalaPongActor extends Actor {
  override def receive: Receive = {
    case "Ping" => sender() ! "Pong"
    case _ => sender() ! Status.Failure(new Exception("unknown message"))}}Copy the code

    The above code uses Scala to implement a simple Actor. Most of its APIS are the same as those in Java, but some are different as follows:

    methods meaning
    Receive Override the receive method of the base class in Actor. And returns a PartialFunction. Note that the return type of the receive method is receive. Receive is simply a type defined to represent scala.partialFunction [scala.any, scala.unit].
    Sender Returns a response to the message received, either by an Actor or a request from outside the Actor system.
    ! In Scala, the “! To call the tell method. In Scala, the message sender is passed in implicitly, so we no longer need to explicitly pass in a reference to the message sender. In the tell method “!” There is an implicit ActorRef parameter in the method signature of. If the tell method is called outside Actor, the default value of this parameter is set to noSender. Here is the signature of the method:

    def ! (message: Any)(implicit sender: ActorRef = Actor.noSender): Unit
    ? “?” In Scala, it stands for Ask.
    Failure After receipt of the unknown message, the return akka. Actor. Status. The Failure. The Actor itself never returns Failure on its own under any circumstances (even if the Actor itself is faulty). Therefore, if we want to notify the sender of an error, we must actively send them a Failure. Sending back Failure causes the requester’s Future to be marked as a Failure.

    Also note that in Scala there is an implicit variable self in Actor, which gets the value of the message sender. So the message sender of the tell method in Actor is always self.

    implicit final val self = context.self
Copy the code

Pay attention to the public number data artisan record, focus on the field of big data offline, real-time technology dry goods regular sharing! Personal website www.lllpan.top