Akka Streams：如何实现新源生成时终止旧源的flatMapLatest功能

你要的功能正好对应Akka Stream里的flatMapLatest操作符，先纠正下你代码里的拼写错误——应该是flatMapLatest而非flatMatLatest。它的行为完全匹配你的需求：

• 上游AddressSource产出新地址时，立刻终止当前运行的旧地址子流
• 同时启动新地址对应的StreamingSource子流
• 下游只会接收最新子流的输出元素

正确实现代码 #

AddressSource
  .flatMapLatest(address => StreamingSource.from(address))
  .to(Sink...)

为什么flatMapConcat不符合需求 #

flatMapConcat的逻辑是必须等前一个子流完全结束后，才会启动下一个子流，这和你要求的「收到新地址就立刻终止旧Source」的核心需求冲突，所以不适用。

额外注意事项 #

如果旧子流终止时需要清理资源（比如关闭服务连接、释放句柄），可以在StreamingSource的定义中添加生命周期钩子，比如通过watchTermination绑定清理逻辑，确保资源能正确释放。

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