Akka Streams：元素处理间隔超时触发流失败的实现方法咨询

I get it—Akka Streams' built-in timeouts like idleTimeout or completionTimeout don't quite hit the mark here. Those are designed for overall stream idle time or total completion time, not the specific gap between finishing one element's processing and starting the next. Let's walk through two solid approaches to solve this.

Approach 1: Combine Existing Operators (No Custom Stage Needed) #

If you prefer sticking to standard Akka Streams operators, you can chain together mapAsync, zip, and stream splitting to track the timing between processing end and next element start. This works best when your processing is sequential (parallelism = 1), since we need strict element order to track gaps correctly.

Here's a clean implementation:

import akka.stream.scaladsl.{Flow, Source, Sink}
import java.time.{Duration, Instant}
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

// Define your timeout threshold (e.g., 5 seconds)
val interProcessingThreshold = Duration.ofSeconds(5)

// Your core processing logic (replace with your actual work)
val elementProcessing: String => Future[String] = elem => Future {
  Thread.sleep(100) // Simulate processing time
  s"Processed: $elem"
}

// Flow that processes elements and captures completion times
val processedWithCompletion = Flow[String].mapAsync(1)(elementProcessing).map(elem => (elem, Instant.now()))

// Flow that checks time gaps between previous completion and next element start
val timeoutCheckFlow = Flow[String]
  // Split into two branches: raw elements and processed elements with completion times
  .alsoTo(processedWithCompletion.map(_._2).drop(1).to(Sink.foreach { prevCompletionTime =>
    val now = Instant.now()
    val elapsed = Duration.between(prevCompletionTime, now)
    if (elapsed > interProcessingThreshold) {
      throw new RuntimeException(s"Inter-processing timeout exceeded: $elapsed > $interProcessingThreshold")
    }
  }))
  .via(Flow[String].mapAsync(1)(elementProcessing))

// Example usage
Source(List("elem1", "elem2", "elem3"))
  .via(timeoutCheckFlow)
  .to(Sink.foreach(println))
  .run()

How it works: #

1. We split the stream to track two things: raw elements (before processing) and the exact time each element finishes processing.
2. For every element after the first, we compare the time we start processing it against the completion time of the previous element.
3. If the gap exceeds your threshold, we throw an exception to fail the stream immediately.

Approach 2: Custom GraphStage (More Flexible) #

If you want tighter control or need to integrate timeout checks directly with your processing logic, a custom GraphStage is the way to go. This lets you track state (last processing completion time) within the stream stage and perform checks exactly when you need them.

Here's a complete, reusable stage:

import akka.stream._
import akka.stream.stage._
import java.time.{Duration, Instant}
import scala.concurrent.{Future, Promise}
import scala.concurrent.ExecutionContext.Implicits.global

class InterProcessingTimeoutStage[T](threshold: Duration, processingFunc: T => Future[T]) 
  extends GraphStage[FlowShape[T, T]] {

  val in: Inlet[T] = Inlet[T]("InterProcessingTimeoutStage.in")
  val out: Outlet[T] = Outlet[T]("InterProcessingTimeoutStage.out")
  override val shape: FlowShape[T, T] = FlowShape(in, out)

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = 
    new GraphStageLogic(shape) {
      
      private var lastCompletionTime: Option[Instant] = None
      private var isProcessing = false

      setHandler(in, new InHandler {
        override def onPush(): Unit = {
          if (isProcessing) {
            hold(in) // Wait for current processing to finish (sequential only)
          } else {
            val element = grab(in)
            val now = Instant.now()

            // Check timeout against last processing completion
            lastCompletionTime.foreach { time =>
              val elapsed = Duration.between(time, now)
              if (elapsed > threshold) {
                failStage(new RuntimeException(s"Inter-processing timeout: $elapsed > $threshold"))
                return
              }
            }

            // Start processing the element
            isProcessing = true
            processingFunc(element).onComplete { result =>
              getAsyncCallback[T](processedElem => {
                lastCompletionTime = Some(Instant.now())
                isProcessing = false
                push(out, processedElem)
                if (isAvailable(out)) pull(in) // Request next element if downstream is ready
              }).invoke(result.get)
            }
          }
        }
      })

      setHandler(out, new OutHandler {
        override def onPull(): Unit = {
          if (!isProcessing && !hasBeenPulled(in)) pull(in)
        }
      })
    }
}

// Usage example
val processingFunc: String => Future[String] = elem => Future {
  Thread.sleep(100) // Simulate processing
  s"Processed: $elem"
}

val timeoutStage = new InterProcessingTimeoutStage[String](Duration.ofSeconds(5), processingFunc)
Source(List("elem1", "elem2", "elem3"))
  .via(timeoutStage)
  .to(Sink.foreach(println))
  .run()

Key features of this stage: #

• Tracks the exact time each element finishes processing.
• Checks the time gap before starting to process a new element—fails the stream immediately if the threshold is exceeded.
• Handles backpressure correctly by holding elements while processing is in progress.
• Integrates your async processing logic directly into the stage for a clean, cohesive flow.

Important Notes #

• Sequential Processing: Both approaches assume sequential processing (parallelism = 1). If you use parallel processing, element completion order isn't guaranteed, so tracking "finish of X to start of Y" doesn't make logical sense.
• Error Handling: Throwing an exception will fail the entire stream, which aligns with your requirement. Make sure to add error handling (like recoverWithRetries) if you need to restart or mitigate failures.
• Time Precision: Using Instant.now() gives millisecond (or better) precision, which is sufficient for most use cases.

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