《流处理》专题

我有这样的想法： 我怎么能把它写在一个流中呢？收集f.e.

我有一个通用的Streams API问题，我想“高效地”解决。假设我有一个（可能非常大，可能无限）流。我想以某种方式对其进行预处理，例如，过滤掉一些项目，并对一些项目进行变异。让我们假设这个预处理是复杂的，时间和计算密集型的，所以我不想做两次。 接下来，我想对项序列执行两组不同的操作，并使用不同的流类型构造处理每个不同序列的远端。对于无限流，这将是一个forEach，对于有限流，它可能是一个收集器

我很难理解GCP数据流/Apache Beam和Spring Cloud数据流之间的差异。我试图做的是转向一个更云原生的解决方案，用于流数据处理，这样我们的开发人员可以更专注于开发核心逻辑，而不是管理基础设施。 我们有一个现有的流解决方案，由Spring云数据流“模块”组成，我们可以独立迭代和部署，就像微服务一样，效果很好，但我们希望迁移到我们的业务提供的GCP现有平台，要求我们使用GCP数据流。

我是Kafka流媒体的新手。我使用python设置了一个twitter监听器，它运行在localhost:9092kafka服务器中。我可以使用kafka客户端工具（conduktor）并使用命令“bin/kafka-console-consumer.sh--bootstrap-server localhost:9092-topic twitter--from-begind”来使用侦听器生成的流，

我有一连串的弦和空值 我想将它简化为另一个流，其中任何非空字符串序列连接在一起，即像 我发现的第一种方法是创建收集器，首先将完整的输入流减少到具有所有连接字符串列表的单个对象，然后从中创建新流： 但在这种情况下，在任何使用前，如果str2，甚至作为str2。findFirst（），将完全处理输入流。它需要耗费时间和内存的操作，并且在来自某个生成器的无限流上，它将根本不工作 另一种方法-创建将保持中

我有2个数据流，我将其连接并输入到一个CoFlatMap函数中。我需要能够在两个不同的数据流上测试生成消息，但在消息到达时进行协调。在Flink如何做到这一点？

我试图理解反应流和反应流之间的区别，特别是在RxJava的上下文中？ 我所能理解的最多的是，反应流在规范中有一些背压的概念，但在RxJava/Reactive中已经存在了请求（n）接口。 我不介意ELI5的回答。

假设有一个 类如下所示： 现在我有一个< code>Person元素的< code>Stream，该流可能包含多个< code>Person实例，这些实例具有相同的< code>id，但具有不同的< code>discriminator值，即< code>[Person{"id": 1，" discriminator": "A"}，Person{"id": 1，" discriminator":

我想复制一个Java8流，这样我可以处理它两次。我可以

我想复制一个Java 8流，这样我可以处理它两次。我可以作为列表并从中获取新流； 但我觉得应该有一个更高效/更优雅的方法。 有没有一种方法可以复制流而不将其变成集合？ 我实际上正在处理一个流，所以在移动到右侧投影并以另一种方式处理之前，希望先以一种方式处理左侧投影。类似于这样（到目前为止，我不得不使用技巧）。

我有一个输入流，其中包含XML数据，我想在返回输入流之前对其进行解析。 当然，我可以将流保存为字节数组，并从中返回一个新的InputStream，或者 在“myObj”上创建第二个InputStream。 但是有什么方法可以“即时”解析流吗？ 编辑: 基本上，我正在寻找一种在解析后重用流的方法。在不消耗流的情况下解析流，分别在解析后重置流。 解决方案： 我找到的解决方案是使用BufferedInp

正如您所看到的，在平面映射之后，我应该得到从到的连续数字的有序流。我拆分了一次拆分器，所以它应该跳到某个中间位置。接下来，我从它中消耗一个元素，并再拆分一次。之后，我打印所有剩余的元素。我希望我将有几个连续的元素从流尾（可能零元素，也会很好）。然而，我得到的是和，然后突然跳转到。 我知道目前在JDK中拆分器不是这样使用的：它们总是在遍历之前拆分。但是，官方文档并没有明确禁止在之后调用。 当我使用直

上周在一条流中出现了非常奇怪的NPE，这给我带来了很多麻烦，所以现在我觉得在使用流时使用NPE太安全了。 下面是我现在的方法： 我的问题是，我在这里处理的是外部POJO，所以我无法更改它并使其为空安全，所以我必须调整我的代码。 这里有一些限制：1）errorList-此处不能为null，因此调用是安全的-当它为空时，它只会返回false 2）和都可以为null，这就是为什么我使用这样的过滤器来确保