问题标签 [kotlin-coroutine-channel]

今天一直被 Kotlin 流/通道的问题所困扰。本质上，我想获取从流中发出的值，并立即将它们发送到通道中。然后，我们通过公开的方法以流的形式订阅该频道。这里的用例是拥有一个始终在线的频道订阅和一个可以独立打开和关闭的流。

我在这里的简单用例是一个返回 channelFlow 的轮询器。理想情况下，我可以在 collect 方法中向通道发射。这似乎不起作用。我的菜鸟协程的想法是，因为收集和发送正在暂停，所以排放在收集中被暂停，我们被卡住了。

是否有任何用于流或通道的内置函数可以处理此行为或以任何其他方式实现此行为？

这是使用广播通道的事件总线实现。（不允许使用 RxJava2 :( ）

这是我的测试代码。

该测试效果很好，但我想删除延迟（500），但希望测试能够正常工作。如果我现在删除延迟（500），输出是

或者

我的实际需求是，在实际项目场景中，我希望在订阅后发布数据，同样，如果没有订阅者，则需要删除事件。因此，如果在发布调用之前有订阅调用（与调度程序无关），订阅者必须接收所有事件。

我尝试使用相同的范围/调度程序，但没有任何效果。

通道有两个功能允许我们向其中发送事件。 Send和offer。

我想更好地了解两者之间的区别。

我有一些陈述我想检查是真的。

• Send是一个挂起函数。什么会让我的代码（不是线程）等待它完成。所以它在里面的事件send完成/取消后继续运行。或者它只会暂停，直到我可以将事件排队/接收它？
• 这意味着，如果我send从一个通道到另一个通道，第一个通道将被阻塞，直到第二个通道可以接收/排队？
• 如果我有一个Rendezvous Channel 并且它已经在运行某些东西（例如，在挂起时，等待 API）并且我offer是一个新的偶数。这会offer引发异常吗？导致频道收不到？

如果您知道任何其他主要区别，我将很高兴知道。

提前致谢

这是我的 FirebaseOTPVerificationOperation 类，其中定义了我的 MutableStateFlow 属性，并更改了值，

这是我的视图模式，我从中监听状态流属性的变化，如下所示，

问题是

在第一次调用时在 viewmodal 中触发，

但对于第二次调用：

firebasePhoneVerificationListener.phoneVerificationFailed在 viewmodal 中没有被解雇，我不知道为什么会发生，任何原因或解释都会非常合适。

电流输出：

预期输出：

我正在第一次使用 Kotlin Coroutines 从事一个爱好项目。我已经阅读并观看了有关它的视频，我有点了解这个概念。但我遇到了一个问题。让我告诉你我的代码。

所以。我有两个频道。第一个通道是snapshotsChannel，这是Snapshot将到达的地方。我想缓冲Snapshot并且每当一分钟过去时，我想将缓冲区发送到composeSnapshots通道以进行进一步处理。

基本上我得到了很多，Snapshot我不想直接将它们发送到进一步处理。所以这就是为什么我想每对每分钟缓冲它们。

问题出现在startComposing函数中。这fixedRateTimer不是一个可暂停的功能，所以我不能在这里使用发送功能。我现在有点卡住了，因为我找不到解决方案。我研究了 TickerChannel 和 Kotlin Flow，但这似乎不是解决我问题的正确方法。

你知道解决办法吗？

我有一个库，它通过从不同来源读取数据产生值，并将它们提供给库用户可以从中收集的流

为此，我使用此代码

用户可以做这样的事情来获取数据

这按预期工作，当用户启动作业时，我的图书馆可能会继续发送数据并且图书馆的用户将继续接收它们

但是，如果用户取消lastJob并在几秒钟后再次启动它，他们将丢失我的库在这两次之间接收到的数据

是否可以dataChannel存储在该时间跨度内接收到的所有读数offer（直到其构造函数中设置的缓冲级别 10）以及何时dataFlow再次激活以同时发出这些值？

以下代码仅打印 10000 即仅最后一个元素

代码可以在操场上运行。

但是由于 Flow 在单独的调度线程中启动，并且值被发送到 Channel 并且由于 Flow 具有无限的缓冲区，所以它应该接收每个元素直到调用 onEach。但是为什么只有最后一个元素能够被接收？

这是预期的行为还是一些错误？如果它的预期行为有人会尝试仅将最新元素推送到流中，但所有具有特定缓冲区的流都可以接收该元素。

迁移到 Kotlin 1.5 后，我偶然发现了 Channel 的新trySendBlocking()方法存在问题。

所以，我们有一个演员返回 a SendChannel<Command>，Command作为我们自己的数据类。

在 Kotlin 1.4 中向 actor 发送命令时，我们曾经像这样使用它的通道：

现在，建议在 Kotlin 1.5 中使用以下内容：

很好，但是我想知道为什么不使用trySend()而不是trySendBlocking()？有些事情让我摸不着头脑：

1. 如果我们确定通道的容量足够大，并且参与者消耗它的速度足够快，那么通道就不会被填满，并且trySend()总是会成功。我们可以做这样的事情来解决通道可能已满的少数情况：

   do { val result = channel.trySend(command) } while (result.isFailure)

2. 内部trySendBlocking()使用runBlocking()阻塞整个线程。根据文档，runblocking()“旨在将常规阻塞代码与以挂起样式编写的库连接起来，用于主要功能和测试”。但是这里它的使用方式超出了这个记录的用例。它可能会阻塞整个线程。我会认为这是一个坏主意？

3. trySend()线程安全吗？如果多个线程trySend()在同一个通道上同时调用，会不会有麻烦？这就是我们喜欢的原因trySendBlocking()吗？

谢谢

我想在我的一些服务上使用线程安全计数器来计算并发工作。

比如说，有一个 http 服务可以为多个请求提供服务，并且有一个isWorking稍后用于显示微调器的属性。

这是我对计数器的实现：

所以问题是有时对通道的最后一次发送调用没有到达consumeAsFlow代码的一部分。

这是发生的事情的示例日志：

在这里，invoke 被调用一次，true并且有一行表示counter got a 1对应于该真实（增量）调用。但也有一个调用，false我希望有一个相应的counter got a 0行。但是那个永远不会出现。

for c in channel如果这就是你的想法，我也尝试过迭代频道。