java - Akka 流 GraphStage

Question

在 Akka Streams 中建议使用 GraphStage，但我找不到任何有关在 Java 中使用 getStageActor() 方法的文档(我找到的所有文档都使用了 Scala)。

如何将以下代码转换为 Java？

lazy val self: StageActor = getStageActor(onMessage)

和

private def onMessage(x: (ActorRef, Any)): Unit =
{
  x match {
    case (_, msg: String) =>
      log.info("received msg, queueing: {} ", msg)
      messages = messages.enqueue(msg)
      pump()
  }
}

Answer 1

根据 getStageActor method documentation ，它接受类型的值

scala.Function1<scala.Tuple2<ActorRef,java.lang.Object>, scala.runtime.BoxedUnit>

在 Scala 中看起来像

((ActorRef, AnyRef)) => Unit

在 Java 中，这种类型在语义上等同于(使用 Function 接口(interface))

Function<Tuple<ActorRef, Object>, Void>

哪里Tuple<A, B>是一个包含两个 A 类型值的类和B .

因此，调用getStageActor方法中，您需要创建上述类型的值。您可以直接通过构造扩展 AbstractFunction1 的类的实例来完成此操作。 :

import scala.Function1;
import scala.Tuple2;
import scala.runtime.AbstractFunction1;
import scala.runtime.BoxedUnit;

getStateActor(new AbstractFunction1<Tuple2<ActorRef, Object>, BoxedUnit>() {
    @Override
    public BoxedUnit apply(Tuple2<ActorRef, Object> args) {
        return BoxedUnit.UNIT;
    }
});

如果您使用 Java 8，则可以使用 lambda 表达式在语法上更好的方法来实现此目的。

如果您使用 Scala 2.12+，则 scala.Function1是函数式接口(interface)，可以直接使用lambda表达式:

getStateActor((args: Tuple2<ActorRef, Object>) -> BoxedUnit.UNIT);

如果您使用旧版本的 Scala，则由于特征的编译方式，Function1不是一个功能接口(interface)，您需要使用scala-java8-compat图书馆。有了它，代码看起来像

import static scala.compat.java8.JFunction.*;

getStateActor(func((args: Tuple2<ActorRef, Object>) -> BoxedUnit.UNIT));

然后，为了实现该函数的逻辑，您可以使用 _1() 访问元组的元素。和_2()方法:

(args: Tuple2<ActorRef, Object>) -> {
    Object msg = args._2();
    if (msg instanceof String) {
        log.info("received msg, queueing: {} ", msg);
        messages = messages.enqueue((String) msg);
        pump();
    }
    return BoxedUnit.UNIT;
}

这是您想要转换的逻辑的直接翻译。