时间戳

首页 > 解决方案 > 如何使用scala读取tcp流

scala - 如何使用scala读取tcp流

问题描述

我有一个在特定端口上生成 tcp 流的 java jar。

我可以运行 using java 命令java -jar runner.jar，这开始在端口 8888 上生成消息流。

当我这样做时，nc -l 8888我可以看到消息。

我想使用 scala 和另一个框架或工具（如 akka、akka-stream）来阅读这个流。

任何人都可以帮助我了解阅读此 tcp 流的最佳工具、框架或任何其他技术。

我尝试使用带有以下代码的akka​​流：-

implicit val system = ActorSystem()
    implicit val mater = ActorMaterializer() val ss  = Tcp().outgoingConnection("127.0.0.1", 8888)
      .to(Sink.foreach(println(_)))
    Source.empty.to(ss).run()

我也试过

Tcp().outgoingConnection(new InetSocketAddress("127.0.0.1", 8888))
        .runWith(Source.maybe[ByteString], Sink.foreach(bs => println(bs.utf8String)))

这行不通。

我只需要自己阅读消息和处理。

谢谢

标签： scalaakkaakka-stream

解决方案

据我了解，您想设置 TCP 服务器，这里是使用 akka 流的 TCP Echo示例

def server(system: ActorSystem, address: String, port: Int): Unit = {
implicit val sys = system
import system.dispatcher
implicit val materializer = ActorMaterializer()

val handler = Sink.foreach[Tcp.IncomingConnection] { conn =>
  println("Client connected from: " + conn.remoteAddress)
  conn handleWith Flow[ByteString]
}

val connections = Tcp().bind(address, port)
val binding = connections.to(handler).run()

binding.onComplete {
  case Success(b) =>
    println("Server started, listening on: " + b.localAddress)
  case Failure(e) =>
    println(s"Server could not bind to $address:$port: ${e.getMessage}")
    system.terminate()
}

}

推荐阅读