问题标签 [genstage]

我找到了一些关于如何测试生产者的资源，但是我找不到任何东西来显示如何测试消费者。

在生产者中，我创建了一个虚拟消费者并且一切正常，但是在消费者中我正在努力进行测试。

生产者测试：

和消费者：

提前谢谢你。

在https://hexdocs.pm/gen_stage/GenStage.html#module-init-and-subscribe_tosubscribe_to我使用选项定义 GenStage 模块

如果我手动运行它们，它可以工作

然后它建议可以将其添加到主管树中：

在监督树中，这通常是通过启动多个工作人员来完成的：

那是我的主管树，但是在运行应用程序时，iex -S mix我收到：

**（混合）无法启动应用程序testdep：TestDep.Application.start（：正常，[]）返回错误：关闭：启动子进程失败：B **（退出）无进程：进程不存在或存在当前没有与给定名称关联的进程，可能是因为它的应用程序没有启动

我的应用程序定义mix.ex为

有什么我想念的吗？

我有一个流产生数据的速度不如消耗的快。

所以我有一个这样定义的生产者：

然后我的生产者消费者订阅它

我的消费者订阅了 mu producer-consumer

我遇到的问题是消费者挂起，我认为是因为在某些时候生产者没有设法获取新数据，并且如文档中所述：

当可枚举完成或停止时，该阶段将以 :normal 原因退出。这意味着，如果消费者订阅了可枚举阶段并且：取消选项设置为：永久，这是默认的，消费者也将退出：正常原因

所以我读了更多，它建议添加cancel:: transient不完成阶段的选项。我像这样添加了它，但它不起作用，我错过了什么吗？

最初我使用的是 aFlow.into_stages(flow, [ProducerConsumer])但我不能这样做，因为我无法从我的主管树中引用（或者我不知道如何） ProducerConsumer

更新

从子定义更新对 Flow.into_stages 的传递引用

**（混合）无法启动应用程序测试：Application.start（：正常，[]）返回错误：关闭：无法启动子：生产者**（退出）退出：GenServer.call（{：name，ProducerConsumer }, {:"$subscribe", nil, #PID<0.2031.0>, [cancel: :transient]}, 5000) ** (EXIT) 没有连接到 Elixir.ProducerConsumer

从我在这里阅读的内容：https ://hexdocs.pm/gen_stage/ConsumerSupervisor.html

ConsumerSupervisor 的所有实现只为每个工作单元启动一个子项（上面链接中的打印机模块）。如果孩子死了，有没有办法让孩子重新开始？

对我来说，如果孩子没有正常关机，它可以使用名称“ConsumerSupervisor”重新启动孩子。有没有人这样做过？

在我的实现中，我让消费者​​主管启动一个实际上是 GenServer 的孩子来执行工作，然后关闭它自己。如果它异常崩溃，我希望它重新启动。

想法..我考虑过只实现一个消费者，它调用一个动态监督器然后启动孩子，但这并不能解释反向压力..

这是我实现它的方式，但希望孩子在崩溃时重新启动：

旋转起来的孩子看起来像这样：

谢谢

我目前正在使用 Elixir 构建一个软实时事件处理系统，现在我试图围绕 GenStage/Flow 来了解这是否是构建的正确抽象。不幸的是，大型应用程序的示例很少，大多数是关于过时数据的并行处理。我使用无限事件流作为源。

我的问题是，在我订阅将事件推送到我的应用程序的无限外部事件流的情况下，在 GenStage/Flow 上构建是否有意义。我希望在事件到达服务器时立即处理它们。那就是我不想缓冲它们，直到我得到 500 让 Flow 开始需求。但是使用 1 的最小需求有意义吗？

是否有可能捕获一个可能的错误，在handle_events该错误中创建此异常？

我尝试在消费者中收听回调def terminate(reason, state) do，但没有被调用。

我想知道 genstage 是否有某种我可以使用的错误处理，而不必包含handle_events在 try/catch 中

作为数据流的一部分，我想以 10 个请求/秒的速度下载文件列表。

像这里一样休眠线程是最好的方法吗？

是否可以做类似gen_stage 这里所做的事情，或者使用速率限制部分的最佳方法是什么Flow？

可以@behavior GenStage.Dispatcher访问其调用他的父模块的自定义调度程序{:producer, state, dispatcher: CustomDispatcher}吗？

我已经意识到调度程序是在与生产者相同的进程中启动的。

例如，如果我尝试handle_call从调度程序向生产者启动一个，那么我将收到：

那么有这个错误，还有其他明显的方法可以访问生产者的方法吗？

我知道您可以使用参数初始化调度程序，但我想访问生产者方法/状态，因为它们是通过生产者回调修改的

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

例外：

这是管道：

GenStage：

有一个带有genstage的项目。

生产者 A、生产者-消费者 B 和消费者 C。

B 在 A 处请求事件一（min_demand：0，max_demand：1）。

C 向 B 请求多个事件（min_demand：25，max_demand：50）。

但实际上，C 一次接收和处理一个数据（我从应用程序日志中看到了这一点）。

如何使C批量处理数据？