草庐IT

scala - Akka Streams TCP 套接字客户端终止

coder 2023-09-19 原文

我有以下流程:

val actorSource = Source.actorRef(10000, OverflowStrategy.dropHead)

val targetSink = Flow[ByteString]
    .map(_.utf8String)
    .via(new JsonStage())
    .map { json =>
      MqttMessages.jsonToObject(json)
    }
    .to(Sink.actorRef(self, "Done"))

  sourceRef = Some(Flow[ByteString]
    .via(conn.flow)
    .to(targetSink)
    .runWith(actorSource))

在 Actor 中(即 Sink.actorRef 之一)。 conn.flow 是使用 Tcp().bind(address, port) 的传入 TCP 连接。

目前 Sink.actorRef Actor 在客户端关闭 tcp 连接时继续运行。有没有办法注册 tcp 连接的客户端终止以关闭 Actor？

编辑: 我尝试按照建议处理这两种情况:

case "Done" =>
  context.stop(self)

case akka.actor.Status.Failure =>
  context.stop(self)

但是当我使用套接字客户端进行测试并取消它时，actor 并没有被关闭。因此，如果 TCP 连接终止，“完成”消息和失败似乎都不会被注册。

完整代码如下:

private var connection: Option[Tcp.IncomingConnection] = None
private var mqttpubsub: Option[ActorRef] = None
private var sourceRef: Option[ActorRef] = None

private val sdcTopic = "out"
private val actorSource = Source.actorRef(10000, OverflowStrategy.dropHead)

implicit private val system = context.system
implicit private val mat = ActorMaterializer.create(context.system)

override def receive: Receive = {

case conn: Tcp.IncomingConnection =>
  connection = Some(conn)

  mqttpubsub = Some(context.actorOf(Props(classOf[MqttPubSub], PSConfig(
    brokerUrl = "tcp://127.0.0.1:1883", //all params is optional except brokerUrl
    userName = null,
    password = null,
    //messages received when disconnected will be stash. Messages isOverdue after stashTimeToLive will be discard
    stashTimeToLive = 1.minute,
    stashCapacity = 100000, //stash messages will be drop first haft elems when reach this size
    reconnectDelayMin = 10.millis, //for fine tuning re-connection logic
    reconnectDelayMax = 30.seconds
  ))))

  val targetSink = Flow[ByteString]
    .alsoTo(Sink.foreach(println))
    .map(_.utf8String)
    .via(new JsonStage())
    .map { json =>
      MqttMessages.jsonToObject(json)
    }
    .to(Sink.actorRef(self, "Done"))

  sourceRef = Some(Flow[ByteString]
    .via(conn.flow)
    .to(targetSink)
    .runWith(actorSource))

case msg: MqttMessages.MqttMessage =>
  processMessage(msg)

case msg: Message =>
  val jsonMsg = JsonParser(msg.payload).asJsObject
  val mqttMsg = MqttMessages.jsonToObject(jsonMsg)

  try {
    sourceRef.foreach(_ ! ByteString(msg.payload))
  } catch {
    case e: Throwable => e.printStackTrace()
  }


case SubscribeAck(Subscribe(topic, self, qos), fail) =>

case "Done" =>
  context.stop(self)

case akka.actor.Status.Failure =>
  context.stop(self)
}

最佳答案

the Actor keeps running

你指的是哪个 actor，你在 Sink.actorRef 注册的那个？如果是，那么要在流关闭时将其关闭，您需要处理其中的 "Done" 和 akka.actor.Status.Failure 消息并调用 context.stop(self) 显式。 "Done" 消息将在流成功关闭时发送，而 Status.Failure 将在出现错误时发送。

有关详细信息，请参阅 Sink.actorRef API 文档，它们解释了终止语义。

关于scala - Akka Streams TCP 套接字客户端终止，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/37591558/

套接字套接codeactorRefsectionscalatcpakka-stream

有关scala - Akka Streams TCP 套接字客户端终止的更多相关文章

  1. ruby - 分布式事务和队列，ruby，erlang，scala - 2

    我有一个涉及多台机器、消息队列和事务的问题。因此，例如用户点击网页，点击将消息发送到另一台机器，该机器将付款添加到用户的帐户。每秒可能有数千次点击。事务的所有方面都应该是容错的。我以前从未遇到过这样的事情，但一些阅读表明这是一个众所周知的问题。所以我的问题。我假设安全的方法是使用两阶段提交，但协议(protocol)是阻塞的，所以我不会获得所需的性能，我是否正确？我通常写Ruby，但似乎Redis之类的数据库和Rescue、RabbitMQ等消息队列系统对我的帮助不大——即使我实现某种两阶段提交，如果Redis崩溃，数据也会丢失，因为它本质上只是内存。所有这些让我开始关注erlang和

  2. 网络编程套接字 - 2

    网络编程套接字网络编程基础知识理解源`IP`地址和目的`IP`地址理解源MAC地址和目的MAC地址认识端口号理解端口号和进程ID理解源端口号和目的端口号认识`TCP`协议认识`UDP`协议网络字节序socket编程接口`sockaddr``UDP`网络程序服务器端代码逻辑：需要用到的接口服务器端代码`udp`客户端代码逻辑`udp`客户端代码`TCP`网络程序服务器代码逻辑多个版本服务器单进程版本多进程版本多线程版本线程池版本服务器端代码客户端代码逻辑客户端代码TCP协议通讯流程TCP协议的客户端/服务器程序流程三次握手（建立连接）数据传输四次挥手（断开连接）TCP和UDP对比网络编程基础知识

  3. ruby - 是否可以在不实际发送或读取数据的情况下查明 ruby​​ 套接字是否处于 ESTABLISHED 或 CLOSE_WAIT 状态？ - 2

    s=Socket.new(Socket::AF_INET,Socket::SOCK_STREAM,0)s.connect(Socket.pack_sockaddr_in('port','hostname'))ssl=OpenSSL::SSL::SSLSocket.new(s,sslcert)ssl.connect从这里开始，如果ssl连接和底层套接字仍然是ESTABLISHED，或者它是否在默认值7200之后进入CLOSE_WAIT，我想检查一个线程几秒钟甚至更糟的是在实际上不需要.write()或.read()的情况下关闭。是用select()、IO.select()还是其他方法完成

  4. ruby-on-rails - Ruby 的 'open_uri' 是否在读取或失败后可靠地关闭套接字？ - 2

    一段时间以来，我一直在使用open_uri下拉ftp路径作为数据源，但突然发现我几乎连续不断地收到“530抱歉，允许的最大客户端数(95)已经连接。”我不确定我的代码是否有问题，或者是否是其他人在访问服务器，不幸的是，我无法真正确定谁有问题。本质上，我正在读取FTPURI:defself.read_uri(uri)beginuri=open(uri).readuri=="Error"?nil:urirescueOpenURI::HTTPErrornilendend我猜我需要在这里添加一些额外的错误处理代码...我想确保我采取一切预防措施来关闭所有连接，这样我的连接就不是问题所在，但是我

  5. ruby - Faye WebSocket，关闭处理程序被触发后重新连接到套接字 - 2

    我有一个super简单的脚本，它几乎包含了FayeWebSocketGitHub页面上用于处理关闭连接的内容:ws=Faye::WebSocket::Client.new(url,nil,:headers=>headers)ws.on:opendo|event|p[:open]#sendpingcommand#sendtestcommand#ws.send({command:'test'}.to_json)endws.on:messagedo|event|#hereistheentrypointfordatacomingfromtheserver.pJSON.parse(event.d

  6. ruby - 在 TCPServer (Ruby) 中，我如何从客户端获取 IP/MAC？ - 2

    我想在Ruby的TCPServer中获取客户端的IP地址。以及(如果可能的话)MAC地址。例如，Ruby中的时间服务器，请参阅评论。tcpserver=TCPServer.new("",80)iftcpserverputs"Listening"loopdosocket=tcpserver.acceptifsocketThread.newdoputs"Connectedfrom"+#HERE!HowcanigettheIPAddressfromtheclient?socket.write(Time.now.to_s)socket.closeendendendend非常感谢!

  7. ruby - 使用 ruby​​ 进行套接字编程是个好主意吗？ - 2

    我选择的语言是Ruby，但因为Twitter，我知道Ruby不能处理很多请求。将它用于套接字开发是个好主意吗？或者我应该像Twitter开发人员那样使用像erlang或haskell或scala这样的函数式语言吗？ 最佳答案 我工作的公司使用Ruby作为我们的网站。到目前为止，我们已经处理了超过34,000,000,000次点击。我们每天处理大约10,000,000次点击没有问题。每天的点击量峰值已超过40,000,000次。可扩展性取决于很多因素。例如，与读取相比，我们的数据库执行的写入比例高得不成比例。虽然大多数网站执行大约90

  8. ruby - 重新连接 tcpsocket(或如何检测已关闭的套接字) - 2

    我有一个连接到服务器的ruby​​tcpsocket客户端。在发送数据之前如何检查套接字是否已连接？我是否尝试“拯救”断开连接的tcpsocket，重新连接然后重新发送？如果是这样，有没有人有一个简单的代码示例，因为我不知道从哪里开始:(我很自豪我设法在rails中获得了一个持久连接的客户端tcpsocket。然后服务器决定杀死客户端，一切都崩溃了；)编辑我已经使用此代码解决了一些问题-如果未连接，它将尝试重新连接，但如果服务器已关闭则不会处理这种情况(它将继续重试)。这是正确方法的开始吗？谢谢defself.write(data)begin@@my_connection.write(

  9. ruby-on-rails - 为什么我必须在使用客户验证器后重新加载 rspec 中的记录？ - 2

    我有一个模型User，它在创建后的回调中创建了选项#Userhas_one:user_optionsafter_create:create_optionsprivatedefcreate_optionsUserOptions.create(user:self)end我对此有一些简单的Rspec覆盖:describe"newuser"doit"createsuser_optionsaftertheuseriscreated"douser=create(:user)user.user_options.shouldbe_kind_of(UserOptions)endend一切正常，直到我将自

  10. ruby-on-rails - Rails 是否支持监听 UDP 套接字的简洁方式？ - 2

    在Rails中，什么是集成更新模型某些元素的UDP监听过程的最佳方式(特别是它将向其中一个表添加行)。简单的答案似乎是在同一个进程中使用UDP套接字对象启动一个线程，但我什至不清楚我应该在哪里做适合Rails方式的事情。有没有一种巧妙的方法来开始收听UDP？具体来说，我希望能够编写一个UDPController并在每个数据报消息上调用一个特定的方法。理想情况下，我希望避免在UDP上使用HTTP(因为它会浪费一些在这种情况下非常宝贵的空间)，但我完全控制消息格式，因此我可以为Rails提供它需要的任何信息。 最佳答案 Rails是一个

随机推荐