如何在 Akka Actor 中使用 MYSQL jdbc 实现答案

【问题标题】：How to implement using MYSQL jdbc in an Akka Actor如何在 Akka Actor 中使用 MYSQL jdbc 实现
【发布时间】：2016-04-07 18:47:39
【问题描述】：

嘿，我读了这个 jdbc 文档 https://www.playframework.com/documentation/2.1.0/ScalaDatabase 这个问题 Is it good to put jdbc operations in actors?

现在我的 mysql 事务有一个 ActorClass，每当请求到来时，这个 Actor 都会实例化几次。所以每个请求都会实例化新的actor。连接池安全吗？

我可以使用

val connection = DB.getConnection()

连接对象可以处理异步事务吗？所以我可以只用一个单例来处理 mysql 连接并在所有实例化的演员中使用它。另外如果我想使用anorm，我该如何创建一个隐式连接对象？

谢谢

【问题讨论】：

为什么要为每个 MySQL 事务使用一个参与者？
我想对所有事务使用 MySQL 的 Actor，所以我可以只异步传递消息

标签： mysql scala jdbc playframework akka

【解决方案1】：

如果您不想屏蔽该演员，您的 DB.getConnection() 应该是 promise[Connection] 或 future[Connection]。（答案末尾的警告）

如果您的DB.getConnection() 是同步的（仅返回没有包装类型的连接），您的actor 将挂起，直到它在处理实际消息时实际从池中获得连接。不管你的数据库是否是单例的，最终它都会到达连接池。

话虽如此，您可以创建参与者来处理消息传递和其他参与者来处理数据库中的持久性，将它们放在不同的线程调度程序中，为数据库密集型提供更多线程。 PlayFramework 也建议这样做。

注意事项：

如果你在actor内部运行futures，你不能确保它会运行的线程/时间，我假设你在这些行中做了一些事情（阅读cmets）

def receive = {
  case aMessage => 
    val aFuture = future(db.getConnection)
    aFuture.map { theConn => //from previous line when you acquire the conn and when you execute the next line
                             //it could pass a long time they run in different threads/time 
                             //that's why you should better create an actor that handles this sync and let
                             //akka do the async part
      theConn.prepareStatemnt(someSQL)
      //omitted code...
    }
}

所以我的建议是

//actor A receives, 
//actor B proccess db (and have multiple instances of this one due to slowness from db)

class ActorA(routerOfB : ActorRef) extends Actor {
  def recieve = {
    case aMessage =>
      routerOfB ! aMessage
  }
}

class ActorB(db : DB) extends Actor {
  def receive = {
    case receive = {
      val conn = db.getConnection //this blocks but we have multiple instances 
                                  //and enforces to run in same thread
      val ps = conn.prepareStatement(someSQL)
    }
  }
}

您需要路由：http://doc.akka.io/docs/akka/2.4.1/scala/routing.html

【讨论】：

如何处理连接关闭？ java.sql.SQLException: Connection is closed!
我改进了答案，请查看。以前方法的问题是，当未来由另一个线程运行时，您的连接可能会关闭。
啊，好点。所以它是在actor内部实例化的，但受限于路由到少数actor实例？
因此您可以使用val workers = context.actorOf(Props[ItemProcessingWorker].withRouter(RoundRobinRouter(100))) 实例化一组actor知更鸟时尚。阅读有关路由和调度程序的更多信息，akka 帮助您隔离组件，您不希望您的消息处理程序与您的 db 处理程序共享线程，db 需要更多线程，因为每次获取 conn 时都会阻塞

【解决方案2】：

据我所知，您无法在 RDBMS 的单个连接上运行多个并发查询（即使在 C-API 中，我也没有看到 mysql 的异步/非阻塞调用的任何资源/参考；）。要同时运行查询，您通常需要多个连接实例。
DB.getConnection 在您有多个连接实例时并不昂贵。使用 DB 最昂贵的领域是运行 sql 查询并等待其响应。
要与您的数据库调用异步，您应该在其他线程中运行它们（而不是在 Akka 或 Play 的主线程池中）； Slick 为您做到。它管理一个线程池并在它们上运行您的数据库调用，然后您的主线程将可用于处理收入请求。然后，您不需要将您的 DB 调用包装在 actor 中以实现异步。

【讨论】：

好点，我用这样的模式管理它： MysqlConn (singleton) > 一次处理一个连接，然后在调用时返回对象连接； ActorMysql(Actor Class) > 每次另一个actor或模块想要存储或检索数据时实例化，在这个actor内部会调用MysqlConn.getConnection；嗯，到现在为止还是靠谱的
可靠；但是并发呢？您的单一连接是您的瓶颈，不要让您同时运行多个查询。

【解决方案3】：

对于更实用的数据库访问方式，我建议查看slick，它有一个很好的 API，可以与普通参与者集成，并与streams 一起进一步发展。

【讨论】：

【解决方案4】：

我认为您应该从池中获取连接并在完成后返回。如果我们为每个参与者建立一个连接，如果该连接断开连接，您可能需要重新初始化它。

对于您可能想尝试的交易

DB.withTransaction { conn =>  // do whatever you need with the connection}

【讨论】：