如果您想对从数据库操作中获得的信息进行处理，可以使用akka streams子流按键分组，然后合并子流：

def databaseOp(input: In): Future[Out] = input match {
  case INSERT => ...
  case UPDATE => ...
  case DELETE => ...
}

val databaseFlow: Flow[In, Out, NotUsed] =
  Flow[In].groupBy(Int.maxValues, _.key).mapAsync(1)(databaseOp).mergeSubstreams

请注意，来自输入源的顺序不会像在中那样保留在输出中 mapAsync ，但对同一个键的所有操作仍将是有序的。

你在找什么 Future.flatMap :

def doSomething: Future[Unit]
def doSomethingElse: Future[Unit]

val result = doSomething.flatMap { _ => doSomethingElse }

它执行第一个函数，然后 Future 满意了，开始第二个。这个 result 是一个新的 Future 当第二次执行的结果满足时，这就完成了。
第一个未来的结果被传递到您赋予的函数中 .flatMap ，因此第二个函数可以依赖于第一个函数的结果。例如：

def getUserID: Future[Int]
def getUser(id: Int): Future[User]

val userName: Future[String] = getUserID.flatMap(getUser).map(_.name)

你也可以把它写成 for-comprehension :

for {
  id <- getUserID
  user <- getUser(id)
} yield user.name

如果你用1的并行度处理事物，它们将以严格的顺序执行，这将解决你的问题。
但这并不有趣。如果您愿意，您可以并行运行不同键的操作—如果对不同键的处理是独立的，那么根据您的描述，这是可能的。要做到这一点，您必须缓冲传入的值，然后重新组合它。让我们看看一些代码：

import monix.reactive.Observable
import scala.concurrent.duration._

import monix.eval.Task

// Your domain logic - I'll use these stubs
trait Event
trait Acknowledgement // whatever your DB functions return, if you need it
def toKey(e: Event): String = ???
def processOne(event: Event): Task[Acknowledgement] = Task.deferFuture {
  event match {
    case _ => ??? // insert/update/delete
  }
}

// Monix Task.traverse is strictly sequential, which is what you need
def processMany(evs: Seq[Event]): Task[Seq[Acknowledgement]] =
  Task.traverse(evs)(processOne)

def processEventStreamInParallel(source: Observable[Event]): Observable[Acknowledgement] =
  source
    // Process a bunch of events, but don't wait too long for whole 100. Fine-tune for your data source
    .bufferTimedAndCounted(2.seconds, 100)
    .concatMap { batch =>
      Observable
        .fromIterable(batch.groupBy(toKey).values) // Standard collection methods FTW
        .mapAsync(3)(processMany) // processing up to 3 different keys in parallel - tho 3 is not necessary, probably depends on your DB throughput
        .flatMap(Observable.fromIterable) // flattening it back
    }

这个 concatMap 这里的操作符将确保您的块也按顺序处理。所以即使有一个缓冲区 key1 -> insert, key1 -> update 另一个有 key1 -> delete ，这不会引起任何问题。在monix中，这与 flatMap ，但在其他rx库中 flatMap 可能是的别名 mergeMap 没有订购保证。
这可以通过 Future 同样，虽然没有标准的“顺序遍历”，所以你必须自己滚动，比如：

def processMany(evs: Seq[Event]): Future[Seq[Acknowledgement]] =
  evs.foldLeft(Future.successful(Vector.empty[Acknowledgement])){ (acksF, ev) =>
    for {
      acks <- acksF
      next <- processOne(ev)
    } yield acks :+ next
  }