AKKA入门

1 Guardian.java

package com.example.demo;

import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import akka.actor.typed.pubsub.Topic;

public class Guardian extends AbstractBehavior<Void> {

    private static ActorRef<Topic.Command<Message>> topic = null;
    private static ActorRef<Message> subscriberActor = null;


    private Guardian(ActorContext<Void> context) {
        super(context);
    }


    public static Behavior<Void> create() {
        return Behaviors.setup(context -> {
            context.spawn(MemberEventLogger.create(), "listener");

            topic = context.spawn(Topic.create(Message.class, "my-topic"), "MyTopic");
            subscriberActor = context.spawn(SubscriberActor.create(),"Subscribe");

            topic.tell(Topic.subscribe(subscriberActor));

            return Behaviors.empty();
        });
    }

    public static void publish(Message message) {
        topic.tell(Topic.publish(message));
        return;
    }

    @Override
    public Receive createReceive() {
        return newReceiveBuilder()
                .build();
    }

}

2 MemberEventLogger.java

package com.example.demo;

import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import akka.cluster.ClusterEvent;
import akka.cluster.typed.Cluster;
import akka.cluster.typed.Subscribe;

public class MemberEventLogger extends AbstractBehavior<ClusterEvent.MemberEvent>  {

    private MemberEventLogger(ActorContext<ClusterEvent.MemberEvent> context) {
        super(context);
    }
    public static Behavior<ClusterEvent.MemberEvent> create() {
        return Behaviors.setup(context -> {
            Cluster cluster = Cluster.get(context.getSystem());

            context.getLog().info("Started [{}], cluster.selfAddress = {})",
                    context.getSystem(),
                    cluster.selfMember().address());

            cluster.subscriptions().tell(new Subscribe<>(context.getSelf(), ClusterEvent.MemberEvent.class));

            return Behaviors.receiveMessage(event -> {
                context.getLog().info("MemberEvent: {}", event);
                return Behaviors.same();
            });
        });
    }

    @Override
    public Receive createReceive() {
        return newReceiveBuilder()
                .build();
    }
}

3 Message.java

package com.example.demo;

public class Message implements MySerializable {
    private String text;

    public Message(){};

    public Message(String text) {
        this.text = text;
    }

    public String getText() {
        return text;
    }
}

4 MySerializable.java

package com.example.demo;

public interface MySerializable {
}

5 SubscriberActor.java

package com.example.demo;

import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class SubscriberActor extends AbstractBehavior<Message> {

    private SubscriberActor(ActorContext<Message> context) {
        super(context);
    }

    public static Behavior<Message> create() {
        return Behaviors.setup(context -> new SubscriberActor(context));
    }

    @Override
    public Receive<Message> createReceive() {
        return newReceiveBuilder()
                .onMessage(Message.class, this::onMsg)
                .build();
    }

    private Behavior<Message> onMsg(Message msg) {
        log.info("ip: {} 收到消息：{}", getContext().getSystem().address().toString(), msg.getText());
        return this;
    }

}

6 HelloWorld.java

package com.example.demo;

import akka.actor.typed.ActorRef;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.Adapter;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.pubsub.Topic;
import akka.cluster.ClusterEvent;
import akka.cluster.typed.Cluster;
import akka.cluster.typed.Subscribe;
import akka.management.cluster.bootstrap.ClusterBootstrap;
import akka.management.scaladsl.AkkaManagement;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

@Slf4j
@Service
public class HelloWorld {

    private ActorSystem system = null;
    private static String remoteAddress = "";

    @PostConstruct
    public void start() throws InterruptedException {
        Config config = ConfigFactory.load("application.conf");
        system = ActorSystem.create(Guardian.create(), "Appka",config);
        remoteAddress = system.address().toString();

        AkkaManagement.get(system).start();
        ClusterBootstrap.get(system).start();

        Thread.sleep(20000);
        Guardian.publish(new Message(remoteAddress));

    }
}

7 application.conf

akka {
  loglevel = "DEBUG"
  actor.provider = cluster

  coordinated-shutdown.exit-jvm = on

  actor.serialization-bindings {
      "com.example.demo.MySerializable" = jackson-json
  }

  cluster {
    shutdown-after-unsuccessful-join-seed-nodes = 60s
    downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider"
  }
}

#management-config
akka.management {
  cluster.bootstrap {
    contact-point-discovery {
      # For the kubernetes API this value is substributed into the %s in pod-label-selector
      service-name = "appka"

      # pick the discovery method you'd like to use:
      discovery-method = kubernetes-api

      required-contact-point-nr = 2
      required-contact-point-nr = ${?REQUIRED_CONTACT_POINT_NR}
    }
  }
}
#management-config

#akka.management {
#  health-checks {
#    readiness-checks {
#      example-ready = "akka.cluster.bootstrap.demo.DemoHealthCheck"
#    }
#  }
#}

8 pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>demo</name>
    <description>demo</description>
    <properties>
        <java.version>1.8</java.version>
        <akka.management.version>1.2.0</akka.management.version>
        <scala.binary.version>2.13</scala.binary.version>
    </properties>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>com.typesafe.akka</groupId>
                <artifactId>akka-bom_${scala.binary.version}</artifactId>
                <version>2.7.0</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>

        <!--akka management-->
        <dependency>
            <groupId>com.lightbend.akka.management</groupId>
            <artifactId>akka-management-cluster-http_${scala.binary.version}</artifactId>
            <version>${akka.management.version}</version>
        </dependency>
        <dependency>
            <groupId>com.lightbend.akka.management</groupId>
            <artifactId>akka-management-cluster-bootstrap_${scala.binary.version}</artifactId>
            <version>${akka.management.version}</version>
        </dependency>
        <dependency>
            <groupId>com.lightbend.akka.discovery</groupId>
            <artifactId>akka-discovery-kubernetes-api_${scala.binary.version}</artifactId>
            <version>${akka.management.version}</version>
        </dependency>

        <!--akka-->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-cluster-typed_${scala.binary.version}</artifactId>
        </dependency>

        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-serialization-jackson_${scala.binary.version}</artifactId>
        </dependency>


        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>RELEASE</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>