Spring Boot集成Akka Stream快速入门Demo

news2024/11/20 3:18:16

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数据的转换。

stream

Akka流的特点

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

2.代码工程

实验目标

熟悉akka stream 相关概念

pom.xml

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

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;

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

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

}

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.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;
    @PostConstruct
    public void run() {
        Source<Integer, NotUsed> source = Source.range(1, 10);
        Sink<Integer, CompletionStage<Done>> sink = Sink.foreach(System.out::println);

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

        graph.run(actorSystem);
    }
}

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