13

问题 当我重新启动/完成/停止流时，旧的消费者不会死/关机：

[INFO ] a.a.RepointableActorRef -
  Message [akka.kafka.KafkaConsumerActor$Internal$Stop$] 
  from Actor[akka://ufo-sightings/deadLetters]
  to Actor[akka://ufo-sightings/system/kafka-consumer-1#1896610594]
  was not delivered. [1] dead letters encountered.

描述 我正在构建一个服务，它接收来自 Kafka 主题的消息并通过 HTTP 请求将消息发送到外部服务。

  1. 与外部服务的连接可能会中断，我的服务需要重试请求。

  2. 此外，如果 Stream 中有错误，则需要重新启动整个流。

  3. 最后，有时我不需要流及其相应的 Kafka 消费者，我想关闭整个流

所以我有一个流：

Consumer.committableSource(customizedSettings, subscriptions)
  .flatMapConcat(sourceFunction)
  .toMat(Sink.ignore)
  .run

发送Http请求sourceFunction

我遵循了新文档中的新 Kafka Consumer Restart 说明

  RestartSource.withBackoff(
      minBackoff = 20.seconds,
      maxBackoff = 5.minutes,
      randomFactor = 0.2 ) { () =>
          Consumer.committableSource(customizedSettings, subscriptions)
            .watchTermination() {
                case (consumerControl, streamComplete) =>
                  logger.info(s" Started Watching Kafka consumer id = ${consumer.id} termination: is shutdown: ${consumerControl.isShutdown}, is f completed: ${streamComplete.isCompleted}")
                  consumerControl.isShutdown.map(_ => logger.info(s"Shutdown of consumer finally happened id = ${consumer.id} at ${DateTime.now}"))
                  streamComplete
                    .flatMap { _ =>
                      consumerControl.shutdown().map(_ -> logger.info(s"3.consumer id = ${consumer.id} SHUTDOWN at ${DateTime.now} GRACEFULLY:CLOSED FROM UPSTREAM"))
                    }
                    .recoverWith {
                      case _ =>
                        consumerControl.shutdown().map(_ -> logger.info(s"3.consumer id = ${consumer.id} SHUTDOWN at ${DateTime.now} ERROR:CLOSED FROM UPSTREAM"))
                    }
             }
            .flatMapConcat(sourceFunction)
      }
      .viaMat(KillSwitches.single)(Keep.right)
      .toMat(Sink.ignore)(Keep.left)
      .run

在一个复杂的 Akka 流中，有一个问题在讨论这个非终止的消费者，但还没有解决方案。

是否有强制终止 Kafka 消费者的解决方法

4

1 回答 1

1

如何将消费者包装在 Actor 中并注册 KillSwitch，请参阅：https ://doc.akka.io/docs/akka/2.5/stream/stream-dynamic.html#dynamic-stream-handling

然后在 Actor postStop 方法中，您可以终止流。通过将 Actor 包装在 BackoffSupervisor 中，您可以获得指数退避。

示例演员：https ://github.com/tradecloud/kafka-akka-extension/blob/master/src/main/scala/nl/tradecloud/kafka/KafkaSubscriberActor.scala#L27

于 2018-05-23T14:20:26.577 回答