创建具有相同执行元的actorpublisher和actorsubscriber

Question

现在我需要处理错误，并通过一个错误处理程序将其推送到不同的kafka队列中。我正在尝试将EsHandler同时用作发布服务器和订阅服务器。我不确定如何包括EsHandler作为中间人而不是Sink。

这是我的代码：

val publisher = Kafka.kafka.consume(topic, "es", new StringDecoder())

val flow = Flow[String].map { elem => JsonConverter.convert(elem.toString()) }

val sink = Sink.actorSubscriber[GenModel](Props(classOf[EsHandler]))

Source(publisher).via(flow).to(sink).run()


class EsHandler extends ActorSubscriber with ActorPublisher[Model] {

  val requestStrategy = WatermarkRequestStrategy(100)

  def receive = {
    case OnNext(msg: Model) =>
      context.actorOf(Props(classOf[EsStorage], self)) ! msg

    case OnError(err: Exception) =>
      context.stop(self)

    case OnComplete =>
      context.stop(self)

    case Response(msg) =>
      if (msg.isError()) onNext(msg.getContent())
  }
}

class ErrorHandler extends ActorSubscriber {

  val requestStrategy = WatermarkRequestStrategy(100)

  def receive = {
    case OnNext(msg: Model) =>
      println(msg)
  }
}

Answer 1

我们强烈建议不要实现您自己的处理器（这是reactive streams规范给“subscriber&&Publisher”起的名字。正确使用它是相当困难的，这就是为什么没有Publisher作为helper特性直接公开的原因。

相反，大多数情况下，您希望使用提供给您的源/接收器（或发布服务器/订阅服务器)，并在这些步骤之间运行操作，如映射/筛选等步骤。

事实上，您可以使用现有的Kafka源和汇实现，它称为reactive-kafka，并由Reactive Streams TCK验证，因此您可以相信它是有效的实现。