【发布时间】:2018-06-20 00:33:22
【问题描述】:
我有一个 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 演员模式是否会匹配 MeterData。我的 props 方法中的 ClassTag 是否足以让我通过类型擦除?有什么建议吗?
【问题讨论】:
标签: scala generics apache-kafka akka akka-stream