elixir - GenStage:如何处理生产者无法提供事件的情况？

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

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

我不知道如何解决这个问题。任何帮助是极大的赞赏。这是我到目前为止:

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。但我觉得它仍然很困惑。任何想法我缺少什么？

最佳答案

在 GenStage.handle_demand/2 的文档中有一个重要的(但不幸的是没有用粗体表示)句子:



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

从文档:



但是，文档缺少示例。 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

关于elixir - GenStage:如何处理生产者无法提供事件的情况？，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/42188022/