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

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

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

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

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

有一个生产者，它由一个值列表初始化。

有生产者-消费者，它从生产者那里接收一个值，用这个值执行一些动作。

并且有一个消费者只是简单地展示了这个价值。

我的应用程序.ex：

启动后，我看到这样一个日志：

就是这样，没有其他事情发生。

据我了解，在 中{: noreply, [head], tail}，我为生产者指定了一个新状态。为什么生产者-消费者不请求此列表中的下一个项目？

我有一个 genstage 应用程序的示例，在它的消费者中，我需要使用与 redis 的连接。而且我不明白我需要如何将此连接传递给handle_events。

如果我写：

然后每次在handle_events函数内部调用连接时，都会创建一个新连接。

我怎么解决这个问题？

出于自我教育的目的，我尝试学习 Elixir 并在GenStage图书馆周围转转。

我阅读了文档并得到了大部分内容，但是我对我的特定领域有几个问题。

我尝试构建一个网络抓取工具，它应该每天启动几次并进行一些抓取和后处理。

第一个问题

所以，我最顶层的生产者是一个Stage发出 HTTP 请求并将它们传递给消费者的。

我如何在这里处理“等待 6 小时”？

我应该只接受需求，但将空事件发送给消费者吗？这听起来像是在浪费 CPU 周期。

也许，GenStage对于此类事件不是正确的方法吗？

第二个问题

有时我需要将事件返回到链中。

• ProducerConsumerA加载页面#n
• ProducerConsumerB解析页面并将页面上找到的项目的事件发送给下一个消费者。ProducerConsumerA但它也应该为下一页发送一个事件（如果结果是分页的）

我在使用 GenStage 的演员表时遇到了麻烦。我正在尝试从我的代码的不同部分将数据提供给 Producer，但是当我尝试进行演员调用时，我收到此错误：

ModuleName.handle_cast/2 中没有匹配的函数子句

代码如下：

我正在尝试实现一个非常简单的 GenStage，以确保在我开始介绍我的特定应用程序逻辑之前理解它并让它工作。我收到来自 Producer 的返回数据错误。数据对我来说是正确的。我读过返回的元组中的第二个值必须是一个列表，而且它似乎是一个。

这是我的 GenStage 实现：

当我做

我收到此错误消息输出：

我正在努力解决的是 a) 理解错误消息的含义，当然还有为什么它会出现。

如果我的 GenStage 的handle_demand/2方法如下所示：

Queue当我的（GenServer）更改/更新时，如何让它“重新运行” ？

我的队列模块看起来像这样：

我的 Genserver 消费者看起来像这样：

如果前一台机器很忙，例如等待任务完成，我想在多台机器上“拆分”Processor任务（即 中的任务）。handle_events/3我该怎么做呢？

我已经阅读了关于分布式任务的指南，我只是不确定如何设计它，以便 1. 它在机器之间拆分，以及 2. 它知道何时拆分，即，当它很忙时：

我知道这涉及到一些这样的安排，但我不知道从哪里开始。有人对这种事情有任何经验吗？

更新

我需要这个的原因Processor.process_document(document)是大约需要 30 秒才能完成，并且一次只能运行一个。添加能够承担一半工作量的第二个节点基本上可以将处理时间减半。

我正在尝试编写一个看起来像这样的流：

模块中的其他函数如下所示：

但我不断收到此错误：

第 18 行是该&cleanup/2行。我在这里做错了什么？感觉就像我没有以某种方式返回正确的值......

我已经使用 ElixirFlow有一段时间了，最​​近我尝试使用Flow和Repo.stream使用并行化我的工作流程：

但它只是不起作用。现在我做了一些研究，偶然发现了 Jose Valim的这条评论Repo.stream，他说基本上不兼容，GenStage我相信它也不兼容Flow（因为它建立在 之上GenStage）。

我的问题是，以前有没有人使用 PSQL 作为无界数据源Flow？

P/S：在上面的同一个 GitHub 线程中，有一个“hack”使用 aGenStage来包装Repo.streamthen 充当生产者，但我正在寻找一种更简化的方法，因为我打算使用Flow而不是GenStage