postgresql - DB 的 Scala future

Question

我正在用 scala/play 和 anorm/postgres 编写代码，以根据用户配置文件生成匹配项。以下代码有效，但我注释掉了导致问题的部分，即 while 循环。我在运行它时注意到前 3 个 Futures 似乎同步工作，但是当我在第四步中检索表中的行数时出现问题。

第四步返回上述插入实际发生之前的计数。据我所知，步骤 1-3 正在同步排队等待 postgres，但是检索计数的调用似乎在前 3 个步骤完成之前返回，这对我来说毫无意义。如果前 3 个步骤按正确顺序排队，为什么第四个步骤不等到插入发生后再返回计数？

当我取消注释 while 循环时，会调用匹配生成和插入函数，直到内存用完，因为返回的计数持续低于所需的阈值。

我知道格式本身不合格，但我的问题不是关于如何编写最优雅的 Scala 代码，而是现在如何让它工作。

def matchGeneration(email:String,itNum:Int) = {
  var currentIterationNumber = itNum
  var numberOfMatches = MatchData.numberOfCurrentMatches(email)
    while(numberOfMatches < 150){
          Thread.sleep(25000)//delay while loop execution time
          generateUsers(email) onComplete {
            case(s) => {
               print(s">>>>>>>>>>>>>>>>>>>>>>>>>>>STEP 1")
               Thread.sleep(5000)//Time for initial user generation to take place
              genDemoMatches(email, currentIterationNumber) onComplete {
                case (s) => {
                  print(s">>>>>>>>>>>>>>>>>>>>>>>>>>>STEP 2")
                  genIntMatches(email,currentIterationNumber) onComplete {
                    case(s) => {
                      print(s">>>>>>>>>>>>>>>>>>>>>>>>>>>STEP 3")
                      genSchoolWorkMatches(email,currentIterationNumber) onComplete {
                        case(s) => {
                          Thread.sleep(10000)
                          print(s">>>>>>>>>>>>>>>>>>>>>>>>>>>STEP 4")
                          incrementNumberOfMatches(email) onComplete {
                            case(s) => {
                              currentIterationNumber+=1
                              println(s"current number of matches: $numberOfMatches")
                              println(s"current Iteration: $currentIterationNumber")
                            }
                          }
                        }
                      }
                    }
                  }
                }
              }
            }
          }
        //}
  }

匹配函数被定义为 future ，例如:

def genSchoolWorkMatches(email:String,currentIterationNumber:Int):Future[Unit]=
Future(genUsersFromSchoolWorkData(email, currentIterationNumber))

genUsersFromSchoolWorkData(email:String) 遵循与其他两个相同的形式。它是一个函数，最初获取用户在其个人资料中填写的所有学校/工作字段(SELECT major FROM school_work where email='$email')并生成一个虚拟用户，其中包含一个email:String 用户共有的那些字段。打印此函数大约需要 30-40 行代码，因此如果需要我可以进一步解释。

我已经编辑了我的代码，到目前为止我发现让它工作的唯一方法是使用 Thread.sleep() 破解它。我认为问题可能在于异常 因为我的 Future 逻辑构造确实按我的预期工作，但问题在于写入发生的时间与读取返回的内容不一致。 numberOfCurrentMatches(email:String) 函数返回匹配的数量，因为它是一个简单的 SELECT count(email) from table where email='$email'。问题是有时在插入 23 个匹配后计数返回 0，然后在第二次迭代后它将返回 46。我假设 onComplete() 将绑定(bind)到用 DB.withConnection()< 定义的底层异常函数 但显然它可能距离太远而无法完成此操作。在这一点上，我不太确定应该研究或进一步查找什么来尝试解决这个问题，而不是编写一种单独的监督函数来返回接近 150 的值。

更新

感谢这里用户的建议，并尝试通过此链接了解 Scala 的文档:Scala Futures and Promises

我更新了我的代码，使其更具可读性和 scala 风格:

  def genMatchOfTypes(email:String,iterationNumber:Int) = {
    genDemoMatches(email,iterationNumber)
    genIntMatches(email,iterationNumber)
    genSchoolWorkMatches(email,iterationNumber)
  }
  def matchGeneration(email:String) = {
  var currentIterationNumber = 0
  var numberOfMatches = MatchData.numberOfCurrentMatches(email)
  while (numberOfMatches < 150) {
    println(s"current number of matches: $numberOfMatches")
    Thread.sleep(30000)
    generateUsers(email)
      .flatMap(users =>  genMatchOfTypes(email,currentIterationNumber))
      .flatMap(matches => incrementNumberOfMatches(email))
      .map{
          result =>
            currentIterationNumber += 1
            println(s"current Iteration2: $currentIterationNumber")
            numberOfMatches = MatchData.numberOfCurrentMatches(email)
            println(s"current number of matches2: $numberOfMatches")
        }
    }

  }

我仍然严重依赖 Thread.sleep(30000) 来提供足够的时间来运行 while 循环，然后再尝试再次循环。它仍然是一个笨拙的 hack。当我取消注释 Thread.sleep()

我在 bash 中的输出如下所示:

users for match generation createdcurrent number of matches: 0
[error] c.MatchDataController - here is the list: jnkj
[error] c.MatchDataController - here is the list: hbhjbjjnkjn
current number of matches: 0
current number of matches: 0
current number of matches: 0
current number of matches: 0
current number of matches: 0

这当然是截断的输出。它像这样一遍又一遍地运行，直到我收到有关打开文件太多的错误并且 JVM/播放服务器完全崩溃。

Answer 1

一种解决方案是使用 Future.traverse 获取已知的迭代次数

暗示

object MatchData {
  def numberOfCurrentMatches(email: String) = ???
}

def generateUsers(email: String): Future[Unit] = ???
def incrementNumberOfMatches(email: String): Future[Int] = ???
def genDemoMatches(email: String, it: Int): Future[Unit] = ???
def genIntMatches(email: String, it: Int): Future[Unit] = ???
def genSchoolWorkMatches(email: String, it: Int): Future[Unit] = ???

你可以这样写代码

def matchGeneration(email: String, itNum: Int) = {
  val numberOfMatches = MatchData.numberOfCurrentMatches(email)
  Future.traverse(Stream.range(itNum, 150 - numberOfMatches + itNum)) { currentIterationNumber => for {
    _ <- generateUsers(email)
    _ = print(s">>>>>>>>>>>>>>>>>>>>>>>>>>>STEP 1")
    _ <- genDemoMatches(email, currentIterationNumber)
    _ = print(s">>>>>>>>>>>>>>>>>>>>>>>>>>>STEP 2")
    _ <- genIntMatches(email, currentIterationNumber)
    _ = print(s">>>>>>>>>>>>>>>>>>>>>>>>>>>STEP 3")
    _ <- genSchoolWorkMatches(email, currentIterationNumber)
    _ = Thread.sleep(15000)
    _ = print(s">>>>>>>>>>>>>>>>>>>>>>>>>>>STEP 4")
    numberOfMatches <- incrementNumberOfMatches(email)
    _ = println(s"current number of matches: $numberOfMatches")
    _ = println(s"current Iteration: $currentIterationNumber")
  } yield ()
  }

更新

如果你敦促每次都检查一些条件，一种方法是使用来自 scalaz library 的很酷的单子(monad)事物。 .它为 scala.Future 定义了 monad，所以我们可以在需要时用 asynchronous 替换单词 monadic

例如，StreamT.unfoldM 可以创建条件单子(monad)(异步)循环，即使我们不需要结果集合的元素，我们仍然可以仅将其用于迭代。

让我们定义你的

def generateAll(email: String, iterationNumber: Int): Future[Unit] = for {
  _ <- generateUsers(email)
  _ <- genDemoMatches(email, iterationNumber)
  _ <- genIntMatches(email, iterationNumber)
  _ <- genSchoolWorkMatches(email, iterationNumber)
} yield ()

然后迭代步骤

def generateStep(email: String, limit: Int)(iterationNumber: Int): Future[Option[(Unit, Int)]] =
  if (MatchData.numberOfCurrentMatches(email) >= limit) Future(None)
  else for {
    _ <- generateAll(email, iterationNumber)
    _ <- incrementNumberOfMatches(email)
    next = iterationNumber + 1
  } yield Some((), next)

现在我们的结果函数简化为

import scalaz._
import scalaz.std.scalaFuture._

def matchGeneration(email: String, itNum: Int): Future[Unit] =
  StreamT.unfoldM(0)(generateStep(email, 150) _).toStream.map(_.force: Unit)

看起来同步方法 MatchData.numberOfCurrentMatches 正在对您在 incrementNumberOfMatches 中的异步修改使用react。请注意，通常它可能会导致灾难性的结果，您可能需要将该状态移动到一些 actor 中。或类似的东西