Akka Streams中divertLeft如何区分处理不同类型Left值？

核心结论 #

组合divertTo的方式完全可行，区分不同类型Left值的思路也非常合理——这是针对不同错误场景做差异化处理的典型需求，你的问题出在谓词编写或流操作顺序上。

问题原因分析 #

你之前尝试两次组合divertTo时所有Left值被忽略，大概率是因为：

1. 第一个divertTo的谓词拦截了所有Left值（比如用了_._1.isLeft），导致后续流中没有剩余Left值可处理；
2. collect的类型匹配语法有误，没能正确提取目标类型的元素。

正确实现方案 #

方案1：多次divertTo精准分流 #

通过类型匹配的谓词，依次分流Error和Exception类型的Left值，最后保留Right值传递给下游。注意分流顺序不影响（因为Error和Exception是Throwable的同级子类），但如果有更细分的子类，要优先处理子类避免被父类拦截。

import akka.stream.scaladsl.{Flow, Graph, Sink}
import akka.stream.{SinkShape, FlowWithContext}

def divertErrors[I, CI, R, CO, Mat1, Mat2](
  exceptionSink: Graph[SinkShape[(Exception, CO)], Mat1],
  errorSink: Graph[SinkShape[(Error, CO)], Mat2]
): FlowWithContext[I, CI, R, CO, (Mat1, Mat2)] = {
  flow.via {
    Flow[(Either[Throwable, R], CO)]
      // 分流Error类型的Left值
      .divertTo(
        Flow[(Either[Throwable, R], CO)]
          .collect { case (Left(e: Error), ctx) => (e, ctx) }
          .to(errorSink),
        { case (Left(_: Error), _) => true } // 精准匹配类型的谓词
      )
      // 分流Exception类型的Left值
      .divertTo(
        Flow[(Either[Throwable, R], CO)]
          .collect { case (Left(e: Exception), ctx) => (e, ctx) }
          .to(exceptionSink),
        { case (Left(_: Exception), _) => true }
      )
      // 保留Right值传递给下游
      .collect { case (Right(result), ctx) => (result, ctx) }
  }
}

方案2：使用partition拆分流 #

如果需要更清晰的分支划分，可以用partition将流拆分为三个分支，分别处理Error、Exception和Right值：

def partitionErrors[I, CI, R, CO, Mat1, Mat2](
  exceptionSink: Graph[SinkShape[(Exception, CO)], Mat1],
  errorSink: Graph[SinkShape[(Error, CO)], Mat2]
): FlowWithContext[I, CI, R, CO, (Mat1, Mat2)] = {
  flow.via {
    val sourceFlow = Flow[(Either[Throwable, R], CO)]
    
    // 先拆分出Error分支
    val (errorFlow, restFlow) = sourceFlow.partition {
      case (Left(_: Error), _) => true
      case _ => false
    }
    errorFlow.collect { case (Left(e: Error), ctx) => (e, ctx) }.to(errorSink)
    
    // 从剩余流中拆分出Exception分支
    val (exceptionFlow, rightFlow) = restFlow.partition {
      case (Left(_: Exception), _) => true
      case _ => false
    }
    exceptionFlow.collect { case (Left(e: Exception), ctx) => (e, ctx) }.to(exceptionSink)
    
    // 最终保留Right值的流
    rightFlow.collect { case (Right(result), ctx) => (result, ctx) }
  }
}

关键语法说明 #

判断Left内容类型的谓词可以直接用模式匹配：

• 匹配Error类型：{ case (Left(e: Error), _) => true }
• 匹配Exception类型：{ case (Left(e: Exception), _) => true }
  这种写法会自动过滤不匹配类型的元素，确保分流精准。

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