Scala 中的嵌套过滤器要么在 Future[Try[Int]] 之上

Posted 2023-03-23

【中文标题】Scala 中的嵌套过滤器要么在 Future[Try[Int]] 之上

【英文标题】：Nested Filter in Scala Either on top of Future[Try[Int]]

【发布时间】：2021-09-22 06:20:41

【问题描述】：

我是 scala 的初学者，我正在尝试在 Either 之上实现过滤器逻辑。现在我有一个函数getTaskId，它返回Future[Try[Int]]，而我的filter logic 是基于Int。现在由于 filter 期望 boolean ，我无法在下面的代码 sn-p 中返回相同的内容。

val records: List[CommittableRecord[Either[Throwable, MyEvent]]] = ???
records.filter  
  (x: CommittableRecord[Either[Throwable,MyEvent]]) => 
    x.value match 
      case Right(event: MyEvent) =>
        getTaskId(event.get("task").get) filter 
            case Success(value)     => value > 1
            case Failure(exception) => false
          
      case Left(_) => false
    

我在返回 Future[Try[Int]] 的函数 getTaskId 上收到 filter 的错误

type mismatch;
 found   : scala.concurrent.Future[scala.util.Try[Int]]
 required: Boolean

所以基本上filter 在Future 之上返回另一个Future 但父filter 期待boolean

非常感谢任何帮助。

【问题讨论】：

问题不够清楚，无法回答。你在过滤List[Either]吗？

你的函数应该返回一个Future[Either[A, B]]而不是Either[A, B]，因为你的函数是异步的。

@TausifSayyad 是的，我是 filtering List[Either]

@curiousguy 你想在过滤结束时得到什么？哪种类型？

@BorisAzanov 这应该是我的过滤结果CommittableRecord[Either[Throwable,MyEvent]]

【参考方案1】：

您在这里遇到了 scala 中的两个困难功能：

大量的语法糖 scala 中的最佳实践是不等待Futures 在你内部的一些业务逻辑中使用：Await.result(future, timeout)。你应该只在宇宙的尽头使用它（在大多数情况下：在你的程序结束时）。

因此，我建议重构您当前的逻辑，从过滤 List[CommittableRecord] 以使结果是非阻塞的 - Future[List[CommittableRecord]] 过滤记录列表。您可以使用这个未来，就像它只是另一个数据容器（如Option[T]），并在您的程序调用阻塞操作结束时，如Await.result。

代码示例：

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future.sequence
import scala.util.Failure, Success, Try

case class Task()

type MyEvent = Map[String, Task]
case class CommittableRecord(value: Either[Throwable, MyEvent])

def getTaskId(task: Task): Future[Try[Int]] = ???

val records: List[CommittableRecord] = List.empty[CommittableRecord]
val result: Future[List[CommittableRecord]] = sequence(
  records.map(
    x =>
      (x.value match 
        case Left(_) => Future(false)
        case Right(value) =>
          getTaskId(value.get("task").get)
            .map 
              case Failure(_) => false
              case Success(id) => id > 1
            
      ).map(_ -> x)
  )
).map(
  idMoreThen1AndRecordList =>
    idMoreThen1AndRecordList.collect 
      case (true, record) => record
    
)

或者，经过一些重构并将 lambda 表达式替换为函数：

def isTaskIdMoreThenOneAndRecord(record: CommittableRecord): Future[(Boolean, CommittableRecord)] =
  (record.value match 
    case Left(_) => Future(false)
    case Right(value) =>
      getTaskId(value.get("task").get)
        .map(tryId => tryId.fold(_ => false, _ > 1))
  ).map(_ -> record)

def filterRecordsWithTaskIdMoreThenOne(
    isMoreOneAndRecordList: List[(Boolean, CommittableRecord)]
): List[CommittableRecord] =
  isMoreOneAndRecordList.collect 
    case (true, record) => record
  

val result: Future[List[CommittableRecord]] =
  sequence(records.map(isTaskIdMoreThenOneAndRecord))
    .map(filterRecordsWithTaskIdMoreThenOne)

因此，结果您将拥有Future[List[CommittableRecord]]，并且您可以使用map 函数处理过滤记录：

result.map((filteredRecords: List[CommittableRecord]) => \*do something with filtered records*\)

或者您可以使用flatMap 组合两个非阻塞操作（例如您的列表和另一个非阻塞函数）。

有用的链接：

在scala documentation 上阅读更多关于未来的信息 article about lambda expressions some best practices 在 Scala 中并发