Akka 容错方法

Question

我正在尝试使用 scala+Akka，并且正在尝试找出容错性。我有一个从主管接收消息并将数据插入数据库的演员。 Supervisor 遇到故障时会重新启动 Actor。

我正在更改 postRestart() 中的连接字符串，以防数据库出现连接问题。现在，只要一个 DB 出现连接问题，actor 就会重新启动并开始将数据插入另一个 DB。

这是一个足够好的方法吗？推荐的方法是什么？

主管：

class SocialSupervisor extends Actor {

    override val supervisorStrategy=OneForOneStrategy(loggingEnabled = false){
    case (e:Exception)=>Restart
    }

    val post_ref=context.actorOf(Props[Post])
    def receive={
            case Get_Feed(feed)=>{
                //get data from feed
                post_ref!Post_Message(posted_by,post)
            }
    }
}

演员：

class Post extends Actor{
  val config1=ConfigFactory.load()
    var config=config1.getConfig("MyApp.db")

    override def postRestart(reason: Throwable) {
        config=config1.getConfig("MyApp.backup_db")
        super.postRestart(reason)
    }

    def insert_data(commented_by:String,comment:String){
            val connection_string=config.getString("url")
                val username=config.getString("username")
                val password=config.getString("password")
                //DB operations
    }

    def receive={
      case Post_Message(posted_by,message)=>{
        insert_data(posted_by, message)
      }
    }
}

Answer 1

我认为您可以对代码进行几项改进，使其更加“容错”。

模块化

您可能应该将您的 insert_data 函数与 Actor 的其余部分分开，以便可以独立于任何 ActorSystem 使用和测试它。您的 Actors 中应该只有很少的代码，receive 方法基本上应该是外部函数的调度程序：

object Post {
  def insert_data(conn : Connection)(commented_by : String, comment : String) = {
    ...
  }
}

您甚至可以更进一步，删除 Connection 依赖项。从您的 Actor 的角度来看，插入只不过是一个接受 PostMessage 并返回有效行更新数的函数：

object Post {
  //returns an Int because Statement.executeUpdate returns an Int
  type DBInserter : Post_Message => Int

您现在可以像以前一样插入数据库连接：

  def insertIntoLiveDB(connFactory : () => Connection) : DBInserter = 
    (postMessage : Post_Message) => {
      val sqlStr = s"INSERT INTO .."
      connFactory().createStatement() executeUpdate sqlStr
    }
  }

或者为了测试目的编写一个从不进行插入的函数：

  //does no inserting
  val neverInsert : DBInserter = (postMessage : Post_Message) => 0
}

现在你的 Actor 几乎没有逻辑了：

class Post(inserter : Post.DBInserter) extends Actor {

  def receive = {
    case pm : Post_Message => inserter(pm)
  }

}

容错

到目前为止，应用程序中“故障”的最大来源是网络，在您的案例中，通过 Connection 到数据库来体现。我们需要一些方法让 Connections 在失败的情况下自动刷新。我们可以使用工厂函数来做到这一点：

def basicConnFactory(timeoutInSecs : Int = 10) = {

  //setup initial connection, not specified in question
  var conn : Connection = ???  

  () => {
     if(conn isValid timeoutInSecs)
       conn
     else {
       conn = ??? //setup backup connection
       conn
     }
  }
}

现在在每次插入时都会测试连接的有效性，并在出现问题时重新建立连接。然后可以使用这个工厂来创建 Actor：

import Post.queryLiveDB
val post_ref = 
  context actorOf (Props[Post], insertIntoLiveDB(basicConnFactory()))

随着您的生产要求越来越严格，您可以修改工厂以使用connection pool...