5

以下场景：GenStage 生产者处理 Twitter 流（使用Stream API和ExTwitter）并向 GenStage 消费者提供一组推文（消费者要求的最大需求）。然后消费者只需打印它们。

以下问题：我正在寻找特定的推文，因此并不总是有新的推文可用。如果 GenStage 生产者返回一个空的事件列表，消费者将停止询问。有关更多信息，请参阅此问题和 José Valims 的回复。

我不确定如何解决这个问题。任何帮助是极大的赞赏。这是我到目前为止所拥有的：

defmodule MyApp.TwitterProducer do
  use GenStage
  alias MyApp.TwitterStream

  def start_link(:ok) do
    GenStage.start_link(__MODULE__, :ok)
  end

  def init(:ok) do
    # This creates a regular Elixir Stream
    # I use this as the state so that not every
    # time the consumer asks for new data
    # a new stream is initiated
    stream = TwitterStream.get_stream
    {:producer, stream}
  end

  def handle_demand(demand, stream) do
    # Take tweets from the stream and 
    # turn them into a list. Then return 
    # them to the consumer
    chunk = Stream.take(stream, demand)
    events = Enum.to_list(chunk)
    {:noreply, events, stream}
  end


  def handle_info(_msg, state) do
    # I as getting an "wrong message" error 
    # before I implemented this function myself
    # It does nothing special to my case
    {:noreply, [], state}
  end

end

defmodule MyApp.TwitterConsumer do
  use GenStage

  def start_link() do
    GenStage.start_link(__MODULE__, :ok)
  end

  def init(:ok) do
    {:consumer, :the_state_does_not_matter}
  end

  def handle_events(events, _from, state) do

    Process.sleep(3000)
    IO.inspect(events)

    # We are a consumer, so we would never emit items.
    {:noreply, [], state}
  end

end

# Let's fire this thing up
{:ok, p} = GenStage.start_link(MyApp.TwitterProducer, :ok, name: MyApp.TwitterProducer)
{:ok, c} = GenStage.start_link(MyApp.TwitterConsumer, :ok, name: MyApp.TwitterConsumer)
GenStage.sync_subscribe(c, to: p, max_demand: 3)

会发生什么：这会运行一段时间，然后停止。据我了解，一旦生产者返回一个空事件列表。

编辑：有趣的是：如果我将需求设置为 1，它会继续运行。但它比直接查询 Twitter Stream API 慢得多。这意味着我收到的推文少了十倍。我的理论是，这是由于重复Stream.take调用而不是仅仅调用Enum.to_list整个流。但我发现它仍然非常混乱。有什么我想念的想法吗？

4

1 回答 1

6

文档中有一个重要的（但不幸的是没有用粗体表示）句子GenStage.handle_demand/2：

生产者必须要么存储需求，要么返回请求的事件。

也就是说，而不是阻止Stream.take一个人可能会明确意识到任务可能正在阻止并处理案例，在这种情况下使用Task.await/2合理的超时收集需求（也许Task.yield/2可以用于更复杂的检查，但这里似乎是矫枉过正。）

从文档中：

如果您不希望任务失败，那么您必须heavy_fun/0以与没有异步调用相同的方式更改代码。例如，要么返回{:ok, val} | :error结果，要么在更极端的情况下，使用try/rescue.

但是，文档缺少示例。OTOH，在这里返回空列表并忘记收集需求可能会更容易：

def handle_demand(demand, stream) do
  try do
    task = Task.async(fn ->
      stream
      |> Stream.take(demand)
      |> Enum.to_list()
    end)
    Task.await(task, 1000) # one sec
  catch
    :exit, {:timeout, {Task, :await, [_, 1000]}} ->
      {:noreply, [], stream}
  else
    events when is_list(events) ->
      {:noreply, events, stream}
  end
end
于 2017-02-13T06:47:12.040 回答