通用类型的 Akka 模式匹配

Question

我有一个这样定义的 Actor 实例：

class KafkaPublisher[T <: KafkaMessage] extends Actor {

  override final def receive = {
    case ProducerStreamActivated(_, stream: SourceQueueWithComplete[T]) =>
      context.become(active(stream))

    case other => println("KafkaPublisher got some unknown message while producing: " + other)
  }

  def active(stream: SourceQueueWithComplete[T]): Receive = super.receive orElse {
    case msg: T =>
      stream.offer(msg)

    case other => println("KafkaPublisher got the unknown message while producing: " + other)
  }
}
object KafkaPublisher {

  def props[T <: KafkaMessage](implicit tag: ClassTag[T]) = Props(new KafkaPublisher[T])
}

其中 KafkaMessage 是一个标记特征，我想发布到 Kafka 的每条消息都会扩展这个标记特征。我有一个主管 Actor，它实际上会根据 Kafka 消息的类型创建此 KafkaPublisher Actor 的实例。例如.,

case class MeterData(id: String, meterReadings: Map[DateTime, String]) extends KafkaMessage

所以在我的 SupervisorActor 中，我创建了一个 PublisherActor 实例，如下所示：

def actorFor(props: Props, actorName: String) = context.actorOf(props, actorName)
val actorRef = actorFor(KafkaPublisher.props[MeterData], "meter-data-actor")

同样，我为给定的 KafkaMessage 类型创建相应的 Akka 流。

所以我的问题是，如果我定义了通用模式匹配，这个 KafkaPublisher actor 模式是否会匹配 MeterData。我的 props 方法中的 ClassTag 是否足以让我通过类型擦除？建议？

Answer 1

泛型 T 上的模式匹配应该有效，当范围内存在隐式 ClassTag[T] 时。因此，在 KafkaPublisher:

中提供和存储 ClassTag 就足够了

import scala.reflect.ClassTag

class KafkaPublisher[T <: KafkaMessage : ClassTag] {
  /* ... */
}

object KafkaPublisher {
  def props[T <: KafkaMessage : ClassTag] = Props(new KafkaPublisher[T])
}

如果类型 T 在 props 调用点是具体的，或者它是通用的但那里有一个隐含的 ClassTag[T]，那么您可以调用 props像往常一样：actorFor(KafkaPublisher.props[MeterData], "meter-data-actor")

通用类型的 Akka 模式匹配

Akka Pattern Match on Generic Type

generics

scala

akka

apache-kafka

akka-stream