0

我正在尝试在 Flow 管道中包含一个 GenStage。但是，这会导致引发异常。如果我理解正确，Flow.through_specs/3将产生多个阶段并相应地划分传入项目。我错过了什么吗？我使用 Flow v0.14.2。

例外：

{:down, {%ArgumentError{message: "the :partition option is required when subscribing to a producer with partition dispatcher"}, [{GenStage.PartitionDispatcher, :subscribe, 3, [file: 'lib/gen_stage/dispatchers/partition_dispatcher.ex', line: 143]}, {GenStage, :dispatcher_callback, 3, [file: 'lib/gen_stage.ex', line: 2160]}, {:gen_server, :try_dispatch, 4, [file: 'gen_server.erl', line: 637]}, {:gen_server, :handle_msg, 6, [file: 'gen_server.erl', line: 711]}, {:proc_lib, :init_p_do_apply, 3, [file: 'proc_lib.erl', line: 249]}]}}
Last message: {:DOWN, #Reference<0.1791475268.948436996.112315>, :process, #PID<0.183.0>, {:down, {%ArgumentError{message: "the :partition option is required when subscribing to a producer with partition dispatcher"}, [{GenStage.PartitionDispatcher, :subscribe, 3, [file: 'lib/gen_stage/dispatchers/partition_dispatcher.ex', line: 143]}, {GenStage, :dispatcher_callback, 3, [file: 'lib/gen_stage.ex', line: 2160]}, {:gen_server, :try_dispatch, 4, [file: 'gen_server.erl', line: 637]}, {:gen_server, :handle_msg, 6, [file: 'gen_server.erl', line: 711]}, {:proc_lib, :init_p_do_apply, 3, [file: 'proc_lib.erl', line: 249]}]}}}
State: {%{}, %{done?: true, producers: %{}, trigger: #Function<2.13930487/3 in Flow.Window.Global.materialize/5>}, {1, 6}, [], #Function<42.60253262/4 in Flow.Materialize.mapper_ops/1>}

这是管道：

specs = [
  {
    {PBFParser.Decompressor, []},
    []
  }
]

PBFParser.Reader.stream("test.osm.pbf")
    |> Stream.drop(1)
    |> Stream.take(1_000)
    |> Flow.from_enumerable(max_demand: 100)
    |> Flow.through_specs(specs, max_demand: 5, stages: 6)
    |> Flow.partition(max_demand: 5, stages: 12)
    |> Flow.map(&PBFParser.Decoder.decode_block/1)
    |> Flow.partition(window: Flow.Window.count(20))
    |> Flow.reduce(fn -> [] end, fn batch, total ->
      [batch | total]
    end)
    |> Flow.emit(:state)
    |> Flow.partition(max_demand: 20, stages: 2)
    |> Flow.each(fn item -> IO.inspect(length(item)) end)
    |> Flow.run()

GenStage：

defmodule PBFParser.Decompressor do
  use GenStage

  def start_link(opts \\ []) do
    GenStage.start_link(__MODULE__, opts)
  end

  def init(_) do
    {:producer_consumer, :zlib.open()}
  end

  def handle_events(events, _from, z) do
    ...
    {:noreply, result, z}
  end
end
4

1 回答 1

0

您的消费者订阅的GenStage.PartitionDispatcher是Flow.partition/2.

您声明两个分区（带有stages: 2）选项。

因此，您的消费者应该有partition: ...传递的选项，如此处所示。

于 2018-08-30T15:45:04.263 回答