我通过websocket客户端将传入的消息从Kafka转发到Web服务器。以下代码显示了我是如何做到的:
import akka.Done
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.StatusCodes
import akka.http.scaladsl.model.ws._
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Flow, Keep, Sink, Source}
import com.typesafe.scalalogging.Logger
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import scala.concurrent.{Future, Promise}
final case class WsGraph(logger: Logger, sink: Sink[Message, Future[Done]])(implicit val system: ActorSystem) {
private implicit val materializer = ActorMaterializer()
private implicit val akka = system.settings.config.getConfig("akka.kafka.consumer")
private implicit val executor = system.dispatcher
private val consumerSetup = system.settings.config.getConfig("kafka.consumer.setup")
private val wsSetup = system.settings.config.getConfig("websocket.setup")
private val consumerSettings: ConsumerSettings[String, String] =
ConsumerSettings(akka, new StringDeserializer, new StringDeserializer)
.withBootstrapServers(consumerSetup.getString("bootStrapServers"))
.withGroupId(consumerSetup.getString("groupId"))
.withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
private val kafkaAsSource: Source[Message, (Consumer.Control, Promise[Option[Message]])] = Consumer
.plainSource(
consumerSettings,
Subscriptions.topics(consumerSetup.getString("topics"))
)
.map(msg => TextMessage(msg.value()))
.concatMat(Source.maybe[Message])(Keep.both)
.mapAsync(Runtime.getRuntime.availableProcessors())(Future(_))
private val socketFlow: Flow[Message, Message, (Consumer.Control, Promise[Option[Message]])] =
Flow.fromSinkAndSourceMat(sink, kafkaAsSource)(Keep.right)
private val (upgradeResponse, (draining, _)) =
Http().singleWebSocketRequest(
WebSocketRequest(wsSetup.getString("server")),
socketFlow)
val create: Future[Either[String, Done]] = upgradeResponse.map { upgrade =>
// just like a regular http request we can access response status which is available via upgrade.response.status
// status code 101 (Switching Protocols) indicates that server support WebSockets
if (upgrade.response.status == StatusCodes.SwitchingProtocols) {
logger.info("Switching protocols")
Right(Done)
} else {
Left(s"Connection failed: ${upgrade.response.status}")
}
}
sys.addShutdownHook {
draining.shutdown()
logger.info("Draining websocket ressource.")
}
}
这里的问题是,如果无法访问网络服务器,上面的演员将被关闭。问题是,如何弄清楚,如果网络服务器不再可用,那么演员应该重新启动并尝试再次连接。
我想你的代码
private val (upgradeResponse, (draining, _)) =
Http().singleWebSocketRequest(
WebSocketRequest(wsSetup.getString("server")),
socketFlow)
有返回类型
(Future[WebSocketUpgradeResponse], T)
因为你只使用upgradeResponse
,即Future[WebSocketUpgradeResponse]
也许你可以尝试使用Recover with Retries重写你的代码
所以你必须更换你的
val create: Future[Either[String, Done]] = upgradeResponse.map { upgrade =>
// just like a regular http request we can access response status which is available via upgrade.response.status
// status code 101 (Switching Protocols) indicates that server support WebSockets
if (upgrade.response.status == StatusCodes.SwitchingProtocols) {
logger.info("Switching protocols")
Right(Done)
} else {
Left(s"Connection failed: ${upgrade.response.status}")
}
}
同
planB = Source.empty
Source.fromFuture(upgradeResponse).recoverWithRetries(3, {
case ex: RuntimeException => logger.error("Error", ex); planB
}).runWith(Sink.ignore).map {upgrade =>
// just like a regular http request we can access response status which is available via upgrade.response.status
// status code 101 (Switching Protocols) indicates that server support WebSockets
if (upgrade.response.status == StatusCodes.SwitchingProtocols) {
logger.info("Switching protocols")
Right(Done)
} else {
Left(s"Connection failed: ${upgrade.response.status}")
}
}
在这里,您可以为RuntimeException
添加异常处理。有关详细信息,请参阅https://doc.akka.io/docs/akka/2.5.5/scala/stream/stream-error.html
我希望这将有所帮助。如有任何错误,请通知我。谢谢