《流处理》专题

在我开始使用Spring Cloud Stream之前，我使用的是Spring-Kafka及其对批量消费和自定义错误处理的支持。请注意这段代码的最后两行: 然而，对于Spring Cloud Stream，我找不到如何配置它。我只能找到这些配置属性: Spring、响铃、水流、kafka.bindings.inputconsumer。autoCommitOffset，启用Dlq 因此，在Sprin

我们在库伯内特斯部署了自卫队。从SCDF UI中，我们可以使用基于Docker的源处理器创建流 应用程序日志显示Tomcat没有初始化，因为没有暴露哪些/执行器endpoint 对问题可能是什么以及如何解决有什么想法吗？ SCDF日志 Skipper日志

我正在使用处理器 API 对状态存储进行一些低级处理。关键是我还需要在存储到商店后写入主题。如何在Spring Cloud Streams Kafka应用程序中完成？

问题内容： 我想处理对象中的列表。我必须确保处理所有元素，以便收到它们。 因此l，我应该使用它吗？ 还是只要不使用并行性就足以使用流？ 问题答案： 你在问错问题。你正在询问而要按顺序处理项目，因此你必须询问订购。如果你有顺序的流并执行保证维持顺序的操作，则该流是并行处理还是顺序处理都没有关系；实施将维持秩序。 有序属性不同于并行与顺序。例如，如果你调用一个同时调用流将是无序在List返回的有序流。

我有一个流（KafkaMSG流到一个主题上），有一个flinkKafka消费者，我注意到一个有趣的行为，我想解决这个问题。 当数据正在流入时，如果它在窗口“完成”之前停止，或者如果数据结束（在几个窗口之后）并且没有到达窗口的末尾，则管道的其余部分不会触发。 流程示例： 我正在使用的Flink Kafka消费者010与env时间特性设置为事件时间。和consumer.assign时间戳和水印（新周期

我想在Java对象中处理列表。我必须确保处理所有的元素，以便我收到他们。 因此，我是否应该对我使用的每个调用？ 或者，只要不使用并行性，只使用流就足够了吗？

问题内容： 我使用的是SQL Server 2005 Service Pack2（SP2）（v9.0.3042），那里发布的解决方案对我不起作用。我尝试使用两个连接字符串。我的代码中有一个被注释掉了。 我意识到我可以将所有结果存储在内存中的List或ArrayList中，然后将其返回。我已经成功地做到了，但这不是这里的目标。目标是能够在结果可用时流式传输。 使用我的SQL Server版本可以吗？

我对Spring和Activiti完全陌生，并为自己做了一个运行良好的小项目。该服务中有4个服务任务、一个REST控制器、1个进程、1个服务和4个方法。 当我调用服务器endpoint时，我启动了我的流程，它只是一步一步地完成我的服务任务并调用服务。方法，如表达式${service.myMethod（）}中定义的。 但是，我真正需要的是一个工作流，它在servicecall之后停止，并等待发送另一

我正在运行一个spark作业，流上下文每60秒运行一次。问题是一批处理时间太长（由于计算和保存RDD和Parquet到云存储），一批无法在1分钟内完成。它结束于下一批继续进入并成为活动的（状态=处理）。过了一段时间，我有10个活动批处理，而第一个已经完成。结果，它明显减慢，没有一批能够完成。是否存在严格限制一次活动批处理的数量为1。 多谢了。

我们正在应用程序中使用apache kafka streams 0.10.2.0。我们利用kafka streams拓扑将处理后的数据传递到下一个主题，直到处理结束。 此外，我们使用AWS ECS容器来部署消费者应用程序。我们观察到消费者正在拾取非常旧的消息进行处理，尽管它们已经在更早的时候处理过。这个问题在服务扩展/缩减或新部署时随机发生。我知道在消费者重新平衡时，有些消息可以重新处理。但在这种

顺便说一句：我的应用程序是一些REST控制器和一些批处理作业的组合。那么使用云数据流有意义吗？如果没有，那么是否有更好的控制台管理器用于批处理作业（如重新启动、取消作业门户）等？

我正在创建一个android应用程序，它需要一个到服务器的永久TCP连接。 我创建了一个服务，用于建立连接并侦听Inputstream上的传入字节（该服务在后台运行）。 前4个传入字节表示完整消息的消息长度。 在将完整的消息从Inputstream读入单独的缓冲区后，我想在分析消息的单独线程中调用另一个服务/异步任务。（服务应继续侦听更多传入消息）。 Android/Java中是否有一个现有的Me

我们正在努力计算 1 分钟翻滚时间窗口内不同类型的事件的最大并发计数。 这些事件就像传感器数据，这些数据是从我们的桌面代理每分钟收集的，然而，一些代理得到了一个错误的时间戳，比如说，它甚至比现在晚了几个小时。 所以，我的问题是如何处理/删除这些事件，目前我只是应用过滤器（s = 我的第一个问题是，如果我不这样做，我怀疑这个坏的“未来”事件会触发窗口计算，即使是那些不完整的数据窗口 第二个问题是，我

假设您有一个第三方库，它公开了下一个接口。 代码无法编译，因为Function不知道如何处理由domap声明的CheckedException。我想出了两个可能的解决方案。 解决方案#1-包装调用 解决方案#2-编写实用程序方法

Java8集合提供了以流形式获取集合的特性。然而，一旦我们调用stream（）方法，我们就会以stream的形式获得集合的当前内容。如果我的集合在流处理过程中增长怎么办？流上的操作可能会用更多的数据更新集合。有没有一个简单有效的方法来处理这种情况？