如何使用Elixir Flow按Key分区后再按Offset有序合并流

针对你用Elixir Flow构建并行处理管道的需求，这里给你一套完整的解决方案，刚好适配无限流+10秒超时丢弃+按offset有序合并的场景：

核心思路 #

因为是无限流，我们没法等所有事件都处理完再排序，所以采用10秒滚动窗口来批量收集处理完成的事件，同时维护一个全局的「最后输出offset」标记，确保不会重复输出旧事件，并且每次窗口结束时对收集到的事件按offset排序后输出，超时未处理的事件直接丢弃。

完整代码实现 #

# 生成测试用的无限事件流：{offset, %{"id" => random_id}}
stream = Stream.unfold(1, fn i -> 
  offset = i + 1 
  element = {offset, %{"id" => Enum.random(1..10_000)}} 
  {element, offset} 
end)

# 模拟你的业务处理函数（替换成实际逻辑即可）
process_event = fn {offset, data} -> 
  # 模拟随机处理耗时，用来测试超时场景
  :timer.sleep(:rand.uniform(500))
  {offset, Map.put(data, :processed, true)}
end

stream
|> Flow.from_enumerable()
# 按id分区，8个并行阶段，确保同一个id的事件按输入顺序处理
|> Flow.partition(
  key: fn {_, m} -> Map.get(m, "id") end,
  stages: 8,
  preserve_order: true # 显式开启同key事件的顺序保留，默认也是true，但显式写更稳妥
)
# 并行处理每个分区内的事件
|> Flow.map(process_event)
# 配置10秒滚动窗口，超时后丢弃未到达的事件
|> Flow.window(
  :tumbling,
  period: 10_000,          # 每10秒触发一次窗口合并
  allowed_lateness: 0,     # 窗口关闭后不再接受迟到的事件（直接丢弃）
  discard: :lateness       # 丢弃迟到事件的策略
)
# 维护全局状态：记录最后输出的offset，以及当前窗口收集的事件
|> Flow.reduce(
  fn -> %{last_emitted: 0, events: []} end,
  fn event, state ->
    %{state | events: [event | state.events]}
  end
)
# 窗口结束时触发事件合并与排序
|> Flow.emit(:window)
# 处理窗口内的事件：排序、过滤已输出的、更新全局状态
|> Flow.map(fn state ->
  # 按offset升序排序当前窗口的所有事件
  sorted_events = state.events |> Enum.sort_by(fn {offset, _} -> offset end)
  
  # 过滤掉已经输出过的旧事件（offset <= 最后输出的offset）
  new_events = Enum.filter(sorted_events, fn {offset, _} -> offset > state.last_emitted end)
  
  # 更新最后输出的offset为当前窗口中最大的offset
  max_offset = if Enum.empty?(new_events), do: state.last_emitted, else: new_events |> Enum.map(&elem(&1, 0)) |> Enum.max()
  
  # 返回要输出的事件和更新后的状态
  {new_events, %{state | last_emitted: max_offset, events: []}}
end)
# 展开事件列表，得到单个事件的有序流
|> Flow.flat_map(fn {events, _} -> events end)
# 这里替换成你对有序事件的最终处理逻辑
|> Flow.each(fn {offset, data} ->
  IO.puts("✅ 输出有序事件：offset=#{offset}，id=#{data["id"]}")
end)
|> Flow.run()

关键细节说明 #

1. 分区顺序保证：Flow.partition的preserve_order: true确保同一个id的事件会严格按照输入顺序（也就是offset递增的顺序）处理，完全符合你“仅保留每个id的事件顺序”的要求。

2. 超时丢弃机制：通过Flow.window的allowed_lateness: 0和discard: :lateness，任何处理时间超过10秒的事件会被直接丢弃，不会进入合并阶段。

3. 有序合并逻辑：每个窗口结束时，我们会把收集到的所有事件按offset排序，同时用last_emitted标记避免重复输出旧事件，最终输出的流就是严格按offset递增顺序的。

4. 无限流适配：滚动窗口的方式让我们不需要等待整个流结束，每10秒就会处理一批事件，完美适配无限流的场景。

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