Spring Boot集成Akka Stream快速入门Demo

1.什么是Akka Stream？

Akka Streams是一个用于处理和传输元素序列的库。它建立在Akka Actors之上，使流的摄入和处理变得简单。由于它是建立在Akka Actors之上的，它为Akka现有的actor模型提供了一个更高层次的抽象。Akka流由3个主要部分组成--Source、Flow、Sink--任何非循环流至少由2个部分Source、Sink和任意数量的Flow元素组成。这里我们可以说Source和Sink是Flow的特殊情况。这里Flow位于Source和Sink之间，因为它们是应用于Source数据的转换。

Akka流的特点

Akka-streams对于快速流数据非常有用。
它避免了管理角色所需的大量模板代码。
它最适合于基于大数据的应用。
由于它是建立在Akka工具包上的，我们将获得所有Akka工具包的好处，如反应性、分布式、位置透明性、集群、Remoting等。
它提供了可重用性，这意味着一旦我们设计了数据流图，我们就可以重复使用它的任何次数。

2.代码工程

实验目标

熟悉akka stream 相关概念

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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>springboot-demo</artifactId>
        <groupId>com.et</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>akka</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>


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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-autoconfigure</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-actor_2.13</artifactId>
            <version>2.6.0</version>
        </dependency>

        <!-- Akka Streams -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-stream_2.13</artifactId>
            <version>2.6.0</version>
        </dependency>
        <!-- Akka Actor dependency -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-actor-typed_2.13</artifactId>
            <version>2.6.0</version>
        </dependency>

        <!-- Akka Remote dependency -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-remote_2.13</artifactId>
            <version>2.6.0</version>
        </dependency>
        <!-- Akka Cluster dependency -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-cluster-typed_2.13</artifactId>
            <version>2.6.0</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

</project>

config

package com.et.akka.config;

import akka.actor.ActorSystem;
import akka.stream.ActorMaterializer;
import akka.stream.Materializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConditionalOnClass(akka.stream.javadsl.Source.class)
public class AkkaConfig {

  private final ActorSystem system;
  private final ActorMaterializer mat;

  @Autowired
  public AkkaConfig() {
    system = ActorSystem.create("SpringWebAkkaStreamsSystem");
    mat = ActorMaterializer.create(system);
  }

  @Bean
  @ConditionalOnMissingBean(ActorSystem.class)
  public ActorSystem getActorSystem() {
    return system;
  }

  @Bean
  @ConditionalOnMissingBean(Materializer.class)
  public ActorMaterializer getMaterializer() {
    return mat;
  }
}

akka stream

1.源。

这是你的流的入口。每个流中必须至少有一个源。它需要两个类型参数。第一个代表它所发射的数据类型，第二个是它在运行时可以产生的辅助值的类型。如果不产生，我们就使用Akka提供的NotUsed 类型。它只有一个输出点。源可以被认为是发布者。

Source<Integer, NotUsed> source = Source.range(1, 100);

2.Sink :

这是你的流的出口点。每个流中必须至少有一个水槽。Sink 是我们流的最后一个元素。基本上，它是一个由源发送/处理的数据的订阅者。通常它将其输入输出到一些系统IO。它是一个流的终点，因此消耗数据。一个汇有一个单一的输入通道，没有输出通道。当我们想以可重复使用的方式指定数据收集器的行为时，特别需要汇，而且不需要评估流。水槽可以被认为是用户。

Sink<Integer, CompletionStage<Done>> sink = Sink.foreach(System.out::println);

3.流:

流是流中的一个处理步骤。它结合了一个传入通道和一个传出通道，以及通过它的消息的一些转换。如果一个流被连接到一个源，一个新的源就是结果。同样地，一个流连接到一个汇，就会产生一个新的汇。而同时与一个源和一个汇相连的流的结果是RunnableFlow 。因此，它们位于输入和输出通道之间，但只要它们不与源或汇相连，它们本身就不对应于其中一种味道。这里，流位于源和汇之间，因为它们是应用于源数据的转换。

Flow<Integer, Integer, NotUsed> flow = Flow.of(Integer.class).filter(MyStream::isPrime);

4.RunnableGraph :

一个两端分别连接到Source和Sink的Flow可以被运行（），被称为RunnableGraph。即使通过连接所有的源、汇和不同的操作符来构建RunnableGraph，也不会有数据流经它。这就是Materialization的作用!

RunnableGraph<NotUsed> graph = source.to(sink); 
graph.run(actorSystem);

5.Materializer :

Akka流中的流和图就像准备一个蓝图/执行计划。流的物化是将流的描述和分配它所需的所有必要资源的过程，以便运行。这意味着启动处理的Actor，以及根据流的需要，在引擎盖下的更多内容。在运行（物化）RunnableGraph后，我们会得到指定类型的物化值。每个流操作者都可以产生一个物化的值。Akka有.toMat ，以表明我们要转换源和汇的物化值。

source.via(flow).to(sink).run(materializer);

具体类的信息如下：

package com.et.akka.stream;

import akka.Done;
import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.stream.ActorMaterializer;
import akka.stream.javadsl.Flow;
import akka.stream.javadsl.RunnableGraph;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


import javax.annotation.PostConstruct;
import java.util.concurrent.CompletionStage;

@Component
public class MyStream {
    @Autowired
    private ActorSystem actorSystem;
    @Autowired
    private ActorMaterializer materializer;
    @PostConstruct
    public void run() {
        Source<Integer, NotUsed> source = Source.range(1, 100);
        Sink<Integer, CompletionStage<Done>> sink = Sink.foreach(System.out::println);
        Flow<Integer, Integer, NotUsed> flow = Flow.of(Integer.class).filter(MyStream::isPrime);

       /* RunnableGraph<NotUsed> graph = source.to(sink);
        graph.run(actorSystem);*/

        source.via(flow).to(sink).run(materializer);

    }
    // Method to check if a number is prime
    private static boolean isPrime(int number) {
        if (number <= 1) {
            return false;
        }
        if (number == 2) {
            return true;
        }
        for (int i = 2; i < number; i++) {
            if (number % i == 0) {
                return false;
            }
        }
        return true;
    }
}

以上只是一些关键代码，所有代码请参见下面代码仓库

代码仓库

https://github.com/Harries/springboot-demo(akka)

3.测试

启动Spring Boot工程
控制台输出所有质数

正文

Spring Boot集成Akka Stream快速入门Demo

1.什么是Akka Stream？

Akka流的特点

2.代码工程

实验目标

pom.xml

config

akka stream

1.源。

2.Sink :

3.流:

4.RunnableGraph :

5.Materializer :

代码仓库

3.测试

相关阅读

java有没有类似纯粹结果集的dataframe这种比较轻量化的框架?

doris 有没有针对整条数据的清洗处理逻辑啊？

Go 语言中的 JSON 库 V2 版本

doris可以添加正式分区吗？断电几天后，中间少了几个分区

doris一个20万的表与2000万的表join查询，直接查不出来数据，是数据量太大了么

doris的存算分离，有人用过么，效果咋样啊

doris 如何将时间戳转成yyyy-MM-dd啊

Spring Boot集成Akka Cluster快速入门Demo

发表评论取消回复

还没有评论，来说两句吧...

目录[+]

1.什么是Akka Stream？

Akka流的特点

2.代码工程

实验目标

pom.xml

config

akka stream

1.源。

2.Sink :

3.流:

4.RunnableGraph :

5.Materializer :

代码仓库

3.测试

相关阅读

java有没有类似纯粹结果集的dataframe这种比较轻量化的框架?

doris 有没有针对整条数据的清洗处理逻辑啊？

Go 语言中的 JSON 库 V2 版本

doris可以添加正式分区吗？ 断电几天后 ， 中间少了几个分区

doris一个20万的表与2000万的表join查询，直接查不出来数据，是数据量太大了么

doris的存算分离，有人用过么，效果咋样啊

doris 如何 将时间戳转成yyyy-MM-dd啊

Spring Boot集成Akka Cluster快速入门Demo

发表评论取消回复

还没有评论，来说两句吧...

目录[+]

doris可以添加正式分区吗？断电几天后，中间少了几个分区

doris 如何将时间戳转成yyyy-MM-dd啊