Scala中对比两组元组并返回最新元组的Source实现问题

核心逻辑梳理 #

需要完成的核心操作：

• 将异步获取的Future[Map]安全集成到Akka Stream流处理中，避免阻塞
• 按ID匹配两组对象，分别对比ObjectA和ObjectB的version字段，保留版本更高的实例
• 最终输出符合类型要求的Source[(ObjectA, Option[ObjectB]), _]

修正后的完整代码 #

import akka.stream.scaladsl.Source
import scala.concurrent.{ExecutionContext, Future}
import akka.actor.ActorSystem

// 原case class定义（假设已存在）
case class ObjectA(id: String, version: Long)
case class ObjectB(id: String, version: Long)
case class JoinedObject(objectA: ObjectA, objectB: ObjectB)

def keepLatestObjects(joinedSource: Source[JoinedObject, _], futureObjectMap: Future[Map[String, (ObjectA, Some[ObjectB])]])(implicit system: ActorSystem, ec: ExecutionContext): Source[(ObjectA, Option[ObjectB]), _] = {
  // 将Future转换为单元素Source，确保流处理中安全获取全局Map
  Source.future(futureObjectMap).flatMapConcat { objectMap =>
    joinedSource.map { joinedObj =>
      val targetId = joinedObj.objectA.id
      // 按ID从Map中匹配对应条目（处理Map中不存在该ID的情况）
      val storedEntry = objectMap.get(targetId)

      // 对比并保留最新版本的ObjectA
      val latestA = storedEntry match {
        case Some((storedA, _)) if storedA.version > joinedObj.objectA.version => storedA
        case _ => joinedObj.objectA
      }

      // 对比并保留最新版本的ObjectB，包装为Option
      val latestB = storedEntry match {
        case Some((_, Some(storedB))) if storedB.version > joinedObj.objectB.version => Some(storedB)
        case _ => Some(joinedObj.objectB)
      }

      // 返回符合要求的结果元组
      (latestA, latestB)
    }
  }
}

关键修正点说明 #

1. 异步Map的安全集成

   • 避免直接在流中使用Await.result阻塞线程，改用Source.future将Future[Map]转换为流元素，再通过flatMapConcat让后续所有流处理都能访问到这个全局Map
   • 确保整个流处理的异步非阻塞特性

2. ID匹配逻辑修正

   • 利用joinedObj.objectA.id作为匹配Key（题目明确每组元组内ObjectA与ObjectB ID相同，因此无需额外验证ID一致性）
   • 处理Map中不存在对应ID的边界情况：直接保留JoinedObject中的实例

3. 版本对比逻辑

   • 对ObjectA和ObjectB分别做版本判断，仅当存储的实例版本更高时才替换
   • 严格按照需求将ObjectB包装为Option[ObjectB]，即使原始输入是Some[ObjectB]也保持输出类型一致

4. 返回类型修正

   • 确保最终输出的元组类型严格匹配(ObjectA, Option[ObjectB])，解决原代码的类型不匹配问题

内容的提问来源于stack exchange，提问作者Siddharth Shankar