流
- 依赖
- 简介
- Actor Source
- Actor Sink

流

依赖

为了使用 Akka 流类型，你需要将以下依赖添加到你的项目中：

1. <!-- Maven -->
2. <dependency>
3. <groupId>com.typesafe.akka</groupId>
4. <artifactId>akka-stream-typed_2.12</artifactId>
5. <version>2.5.23</version>
6. </dependency>

8. <!-- Gradle -->
9. dependencies {
10. compile group: 'com.typesafe.akka', name: 'akka-stream-typed_2.12', version: '2.5.23'
11. }

13. <!-- sbt -->
14. libraryDependencies += "com.typesafe.akka" %% "akka-stream-typed" % "2.5.23"

简介

「Akka Streams」使对类型安全的消息处理管道建模变得容易。对于类型化的 Actors，可以在不丢失类型信息的情况下将流连接到 Actors。

此模块包含现有ActorRef源的类型化替换，以及「ActorMaterializerFactory」的工厂方法，后者采用类型化ActorSystem。

从这些工厂方法和源可以与来自原始模块的原始 Akka 流构建块混合和匹配。

注释：此模块已准备好用于生产，但仍标记为「可能更改」。这意味着 API 或语义可以在没有警告的情况下进行更改，但这些更改将在 Akka 2.6.0 中收集并执行，而不是在 2.5.x 补丁版本中执行。

Actor Source

发送到特定 Actor 的消息驱动的流可以使用「ActorSource.actorRef」启动。此源具体化为类型化的ActorRef，它只接受与流类型相同的消息。

1. import akka.actor.typed.ActorRef;
2. import akka.japi.JavaPartialFunction;
3. import akka.stream.ActorMaterializer;
4. import akka.stream.OverflowStrategy;
5. import akka.stream.javadsl.Sink;
6. import akka.stream.javadsl.Source;
7. import akka.stream.typed.javadsl.ActorSource;

9. interface Protocol {}

11. class Message implements Protocol {
12. private final String msg;

14. public Message(String msg) {
15. this.msg = msg;
16. }
17. }

19. class Complete implements Protocol {}

21. class Fail implements Protocol {
22. private final Exception ex;

24. public Fail(Exception ex) {
25. this.ex = ex;
26. }
27. }

29. final JavaPartialFunction<Protocol, Throwable> failureMatcher =
30. new JavaPartialFunction<Protocol, Throwable>() {
31. public Throwable apply(Protocol p, boolean isCheck) {
32. if (p instanceof Fail) {
33. return ((Fail) p).ex;
34. } else {
35. throw noMatch();
36. }
37. }
38. };

40. final Source<Protocol, ActorRef<Protocol>> source =
41. ActorSource.actorRef(
42. (m) -> m instanceof Complete, failureMatcher, 8, OverflowStrategy.fail());

44. final ActorRef<Protocol> ref =
45. source
46. .collect(
47. new JavaPartialFunction<Protocol, String>() {
48. public String apply(Protocol p, boolean isCheck) {
49. if (p instanceof Message) {
50. return ((Message) p).msg;
51. } else {
52. throw noMatch();
53. }
54. }
55. })
56. .to(Sink.foreach(System.out::println))
57. .run(mat);

59. ref.tell(new Message("msg1"));
60. // ref.tell("msg2"); Does not compile

Actor Sink

有两个sink可接受类型化的ActorRef。要将流中的所有消息发送给 Actor 而不考虑反压力（backpressure），请使用「ActorSink.actorRef」。

1. import akka.NotUsed;
2. import akka.actor.typed.ActorRef;
3. import akka.stream.ActorMaterializer;
4. import akka.stream.javadsl.Sink;
5. import akka.stream.javadsl.Source;
6. import akka.stream.typed.javadsl.ActorSink;

8. interface Protocol {}

10. class Message implements Protocol {
11. private final String msg;

13. public Message(String msg) {
14. this.msg = msg;
15. }
16. }

18. class Complete implements Protocol {}

20. class Fail implements Protocol {
21. private final Throwable ex;

23. public Fail(Throwable ex) {
24. this.ex = ex;
25. }
26. }

28. final ActorRef<Protocol> actor = null;

30. final Sink<Protocol, NotUsed> sink = ActorSink.actorRef(actor, new Complete(), Fail::new);

32. Source.<Protocol>single(new Message("msg1")).runWith(sink, mat);

为了使 Actor 能够对反压力作出反应，需要在 Actor 和流之间引入一个协议。使用「ActorSink.actorRefWithAck」可以在 Actor 准备接收更多元素时发出需求信号。

1. import akka.NotUsed;
2. import akka.actor.typed.ActorRef;
3. import akka.stream.ActorMaterializer;
4. import akka.stream.javadsl.Sink;
5. import akka.stream.javadsl.Source;
6. import akka.stream.typed.javadsl.ActorSink;

8. class Ack {}

10. interface Protocol {}

12. class Init implements Protocol {
13. private final ActorRef<Ack> ack;

15. public Init(ActorRef<Ack> ack) {
16. this.ack = ack;
17. }
18. }

20. class Message implements Protocol {
21. private final ActorRef<Ack> ackTo;
22. private final String msg;

24. public Message(ActorRef<Ack> ackTo, String msg) {
25. this.ackTo = ackTo;
26. this.msg = msg;
27. }
28. }

30. class Complete implements Protocol {}

32. class Fail implements Protocol {
33. private final Throwable ex;

35. public Fail(Throwable ex) {
36. this.ex = ex;
37. }
38. }

40. final ActorRef<Protocol> actor = null;

42. final Sink<String, NotUsed> sink =
43. ActorSink.actorRefWithAck(
44. actor, Message::new, Init::new, new Ack(), new Complete(), Fail::new);

46. Source.single("msg1").runWith(sink, mat);

英文原文链接：Streams.