Scala stackoverflowexception - 可能的循环依赖?

问题描述 投票:1回答:1

我有类以下的实现

class User(identifier : Int, actor : ActorRef) extends Serializable {
  var userName : String = Random.alphanumeric.take(4 + Random.nextInt(12)).mkString
  var msgRate : Int = 0
  var followers : MutableList[User] = new MutableList[User]()
  var messageQueue = new LinkedBlockingQueue[String](Messages.maxBufferSize)

  override def equals(o : Any) = o match {
    case that : User => that.userName.equals(this.userName)
    case _ => false

  }

  override def hashCode = identifier.hashCode    

  def getRecentMessages(n : Int) : List[String] = {
    var msgList : List[String] = List.empty[String]
    msgList = messageQueue.toArray().toList.asInstanceOf[List[String]]
    return msgList
  }


  def isFollowing(user : User) : Boolean = {
    user.getFollowers().contains(this)
  }

  def isFollowed(user : User) : Boolean = {
      followers.contains(user)
  }

  def getFollowers() : MutableList[User] = {
    return followers
  }

  def addFollower(follower : User) {
    followers += follower
  }

}

当我为一组较小的演员运行时,添加关注者不会导致任何问题,代码运行正常。然而,对于大量演员来说,问题出现了:

java.lang.StackOverflowError
    at akka.actor.SerializedActorRef$.apply(ActorRef.scala:420)
    at akka.actor.LocalActorRef.writeReplace(ActorRef.scala:389)
    at sun.reflect.GeneratedMethodAccessor1.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at java.io.ObjectStreamClass.invokeWriteReplace(ObjectStreamClass.java:1075)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1134)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)

这是以跟随者的形式在类中使用相同的User实例的情况吗?这个问题有解决方案吗?

编辑:根据请求包含更多代码。对于非常大的代码库感到抱歉。让我简单解释一下。我正在开发类似于twitter的消息模拟器,我必须根据用户标记来管理消息。客户端随机生成用户库,以一定速率发送消息,发件人管理数据。

初始握手包括从Interactor()发送整个客户端信息。服务器确认每个客户端,然后定期安排。发送导致stackoverflow异常的整个客户端信息时会出现此问题。在代码中,问题出现在Init in Interactor()的末尾

这是代码:

object ClientApp extends App {

  val system = ActorSystem("TwitterClientActor", ConfigFactory.load("applicationClient.conf"))
  val sActor = system.actorFor("akka.tcp://ServerActor@" + ipAddr + "/user/Server")
  val serverVector = Vector.fill(Messages.nServers)(sActor)
  val serverActor = system.actorOf(Props.empty.withRouter(RoundRobinRouter(routees = serverVector)), "serverRouter")
  val interActor = system.actorOf(Props(new Interactor()))
  var nRequests : Int = 0
  val startTime = java.lang.System.currentTimeMillis()
  interActor ! Init

}

// Intermediate manager system
class Interactor() extends Actor {
  // Member definitions
  import actorSys.dispatcher

  // User list
  for (i <- 0 to clientList.length - 1)
    clientList(i) = new User(i, context.actorOf(Props(new Client(i : Int))))
  readFollowersStats(clientList.length)

  def receive = {

    // Send request to users
    case Init =>
      for (curUser <- clientList)
        serverActor ! RegisterClients(curUser)
    // ISSUE IMMEDIATELY AFTER THIS

    // Schedule after request
    case ScheduleClient(identifier) =>
      if (!limitReached) {
        val curUser = clientList(identifier)
        val cancellable = actorSys.scheduler.schedule(0.milliseconds, curUser.getMsgRate.milliseconds)(sendMsg(curUser))
        cancelMap += (curUser -> cancellable)
      }

    case ClientCompleted =>
      nCompleted += 1
      if (nCompleted == clientList.length) {
            serverActor ! Broadcast(PoisonPill)
            context.system.shutdown()
        }
  }

  def sendMsg(curUser : User) = {
    nMessages.incrementAndGet()

    if (nMessages.get() == Messages.msgLimit) {
      println("Limit reached!")
      limitReached = true
      for (cancellable <- cancelMap.values)
        cancellable.cancel()
    }
    else if (nMessages.get() < Messages.msgLimit) {
      println(nMessages)
      val curSec = java.lang.System.currentTimeMillis()
      val curTime = ((curSec - ClientApp.startTime).toDouble) / 1000
      if (curTime >= Messages.peakStart && curTime < Messages.peakEnd) {
        for (i <- 0 to Messages.peakScale) {
          var rndTweet = randomTweet(curUser)
          curUser.getReference() ! Tweet(rndTweet)
        }
        nMessages.addAndGet(Messages.peakScale - 1)
      }
      else {
        var rndTweet = randomTweet(curUser)
        //println(curUser + " ---> " + rndTweet)
        curUser.getReference() ! Tweet(rndTweet)
      }
    }
  }

  def randomTweet(curUser : User) : String = {
    // Return some random string
  }

  def readFollowersStats(usersCount : Int) {
    // Read the file stats of the format, min-max percentage
        while(file is not empty)
      FollowersGeneration(usersCount, minFollowers.toInt, maxFollowers.toInt, percentage.toDouble)
    }

  }

  def FollowersGeneration(usersCount : Int, minFollowers : Int, maxFollowers : Int, followersPercentage : Double) {

    var noOfFollowers : Int = 0
    var users : Double = (followersPercentage / 100) * usersCount
    var temp : Int = users.toInt
    for (i <- 0 until temp) {
        if (minFollowers < usersCount) {
            // Random follower assignment...
            // CODE ACCESSES FOLLOWERS HERE!!!
            if (!user.isFollowed(clientList(id)))
                user.addFollower(clientList(id))
          }
        }
    }

}

class Client(identifier : Int) extends Actor {
  var serverActor = ClientApp.serverActor
  def receive = {

    case "ACK" =>
      println("Client " + identifier + " activated")
      ClientApp.interActor ! ScheduleClient(identifier)

    case Tweet(tweet) =>
      serverActor ! Tweet(tweet)

      // Other functions


  }

}

EDIT2:关于用例的说明

这是一个客户端 - 服务器模型。客户端:Initiator对象创建Broker类。代理类创建用户角色列表(在这种情况下为类用户)并建立不同用户之间的关系,即分配随机关注者,速率和其他属性。现在,整个用户列表被发送到服务器,服务器在该服务器上激活各个客户端以开始消息传递。客户端现在将随机消息发送到服务器处理它们的服务器。

最初的方法包括使用上面的类User,然后将actorRef存储为类中的成员并将其发送给用户。这是问题,我将用户列表更改为actor类。我必须生成关注者并将其发送到服务器。我传达了代理类,使用消息添加关注者。现在,问题出现在程序化的角度,在服务器端,我必须访问User actor的关注者。我可以发送请求用户的消息'!'或使用'?'获得未来。这是一个会降低服务器处理能力的问题。有更优雅的方法,我可以访问actorRef的成员或更好的解决方案,我可以调用一个函数?

scala linked-list stack-overflow
1个回答
0
投票

上面的代码是将一个actor包装在一个类中的不良实践的一个例子。确保使用actor模型,而不是从面向对象的角度接近它。

在上面的场景中,要定义的结构是:

Broker -> UserActor -> User

Broker有一个map,这可以用来解决它的任何依赖。该模型确保actor模型与对象逻辑分离。 Reference

另一种选择是使用FuturesThis link是一个很好的起点。

© www.soinside.com 2019 - 2024. All rights reserved.