使用akka-http实现web socker服务器时，堆内存不断增加

Question

  def decodeService: Flow[Message, Message, _] = { 
    Flow[Message].map { 
      case BinaryMessage.Strict(encodeMsg) => 
        try {
          WebServer.getWorkerActor ! QueueWork(protoMsg(this, encodeMsg))
        } catch {
          case e: Exception => {
            println("[ERROR] failed to send BinaryMessage.Strict: " + e)
            TextMessage("[ERROR] failed receiving BinaryMessage.Strict")
          }
        }
        TextMessage("[INFO] BinaryMessage.Strict")

      case BinaryMessage.Streamed(streamedMsg) => {
        implicit val system = ActorSystem("DecoderSystem")
        implicit val materializer = ActorMaterializer()
        val streamedMsgFuture: Future[Seq[ByteString]] = streamedMsg.runWith(Sink.seq)
        streamedMsgFuture.onComplete { completedStream =>
          var completeBytestring = new ByteStringBuilder()
          //I'm sure there's a better way to do this.. but hey, it's O(n)
          completedStream.foreach { x => 
            x.foreach { y => 
              completeBytestring ++= y  
            }
          }
          try {
              WebServer.getWorkerActor ! QueueWork(protoMsg(this, completeBytestring.result()))
          } catch {
            case e: Exception => {
              println("[ERROR] failed to send BinaryMessage.Streamed: " + e)
              TextMessage("[ERROR] failed receiving BinaryMessage.Streamed")
            }
          } finally {
            completeBytestring.clear()
          }
        }
        TextMessage("[INFO] BinaryMessage.Streamed")
      }

      case TextMessage.Strict(txt) => TextMessage("Succesfully receive text message")
      case _ => TextMessage("Message type unsupported")
    }
  }

我是否错误地使用了流/接收器/源？如何冲洗溪流？

谢谢

Answer 1

好吧，最突出的问题是，您为接收到的每个流消息创建一个全新的actorSystem。ActorSystem就像您的参与者的线程池；您希望尽可能少地创建它们，理想情况下，整个应用程序只创建一个。您不仅为每个消息创建它们，而且不关闭它们--在ActorSystem中配置的所有分派程序，以及它所持有的所有资源都将永远挂起。当然，如果您接收到大量的流式消息，您的内存使用量将会增加。

由于您使用了akka-http，因此您必须有一个ActorSystem，在此调用http().bind*。您需要在decodeservice方法中使其可访问。此外，您计算组合字节流的方式在我看来似乎过于复杂。考虑这样写：

def decodeService: Flow[Message, Message, _] = Flow[Message].mapAsync(4) {
  case m: BinaryMessage.Strict =>
    Future.successful(m)
  case BinaryMessage.Streamed(streamMsg) =>
    streamMsg.runReduce(_ ++ _).map(BinaryMessage.Strict)
  case m =>
    Future.successful(m)
}.map {
  case BinaryMessage.Strict(encodeMsg) =>
    try {
      WebServer.getWorkerActor ! QueueWork(protoMsg(this, encodeMsg))
      TextMessage("[INFO] BinaryMessage.Strict")
    } catch {
      case NonFatal(e) =>
        println("[ERROR] failed to send BinaryMessage.Strict: " + e)
        TextMessage("[ERROR] failed receiving BinaryMessage.Strict")
    }
  case TextMessage.Strict(txt) => TextMessage("Succesfully receive text message")
  case _ => TextMessage("Message type unsupported")
}

在这里，我们首先将所有二进制消息转换为binarymessage.strict，然后像在原始代码中一样处理它们。注意，您必须在try块内写入确认消息，否则即使出现异常，也会返回成功消息。此外,如果您决定根本不处理文本消息,代码可能会变得更加简单：

def decodeService: Flow[Message, Message, _] = Flow[Message]
  .filterNot(_.isText)
  .mapAsync(4) {
    case BinaryMessage.Strict(binary) =>
      Future.successful(binary)
    case BinaryMessage.Stream(binaryStream) =>
      binaryStream.runReduce(_ ++ _)
  .map { encodeMsg =>
    try {
      WebServer.getWorkerActor ! QueueWork(protoMsg(this, encodeMsg))
      TextMessage("[INFO] BinaryMessage.Strict")
    } catch {
      case NonFatal(e) =>
        println("[ERROR] failed to send BinaryMessage.Strict: " + e)
        TextMessage("[ERROR] failed receiving BinaryMessage.Strict")
    }
  }