问题标签 [monix]

我有一个 Monix Observable，它发出无限的数据流

我想要一个背压控制，它只保留每个 Foo::id 的 1 条记录，并删除旧的：

但显然 100 不是正确的数量，我无法根据 id 字符串精确控制。如果洪水迅速涌入，我可能会得到 99Foo(a,xx)和 1 。Foo(b,xxx)

有什么办法可以做到这一点？

我在 monix 存储库中看到此代码，在使用部分应用技术的对象任务下。我知道当我们想一次推断 1 个类型参数时使用该技术，而在这种情况下推断两者都是不可能的，对吧。为什么在这里使用它？

在使用monix.eval.Task或zio.Task时，是否有一种简单的方法可以将 Task 的 Option 转换为Option的 Task？

CE 2.x。莫尼克斯 3.4。第 1 行编译，第 2 行给出编译错误：

从对象 Eq Stream.sleep_(5.seconds).compile.drain.as(ExitCode.Success) // 2

我正在尝试使用 monix 来并行化某些操作，然后执行错误处理

假设我正在尝试解析和验证几个这样的对象

和

这里 X 和 Y 是我定义的一些类型。

现在，这些方法中的每一个都会评估一些标准并返回一个任务。如果评估失败，我有一些形式的代码

我对 Y 有类似的任务加薪。

现在我有这种类型

我定义了这个

这应该允许我并行运行验证，果然我得到了响应。

但是我也想要这种行为

如果两个任务都失败，我想返回这样的错误

我似乎无法做到这一点。根据两个任务中的哪一个失败，我只能在 parMap2 输出的 onRecover 方法上获得两个失败之一。

如果两者都失败，我只会收到任务 X 的错误。

我是否有可能以完全异步的方式完成我正在使用 Monix 所做的事情（例如，可能是其他一些将任务组合在一起的方法）？还是我必须阻止执行程序，单独获取错误并重新组合值？

我有以下用例。

异步执行 DB 操作，完成后将 kafka 事件发送到另一个微服务，以便它从 DB 中读取。但是，到目前为止，甚至在数据库操作完成之前就已经发送了 kafka 事件。我的代码如下所示：

有什么办法可以确保 pushToKafkaTask 在前三个任务之后肯定发生？

添加更多代码片段以显示 firstTask 、 secondTask 和 pushToKafkaTask 的样子