《stream》专题

我们在Spring Boot应用程序中使用Kafka Cloud Stream向Kafka发送数据。这样地 我想知道除了直接从 yaml 文件中读取之外，是否可以从消息通道获取主题名称？ 主题名称存在于kafka.yaml中

我正在尝试学习java流。我能够进行简单的迭代/过滤/映射/收集等。 当我尝试收集每 3 个元素并打印时，如本例所示，[收集每 3 个元素并打印等等......] 输出： 我没有得到任何线索，如何使用流做到这一点。我应该实现自己的收集器来实现这一点吗？

我正在Jenkins文件中运行以下groovy脚本，该文件在构建期间执行： 上述参数值代码的预期结果等于： 是： 但是，在Jenkins上执行此代码时，我收到一个奇怪的错误，我无法理解（为什么会发生）： hudson.remoting.ProxyExc0019：org.codehaus.groovy.runtime.typehandling.GroovyCastExc0019：无法将对象'Samp

这是一个例子:代码A: 另一个代码B可以这样使用: 两者有什么区别，有和没有？

我想知道这两段代码有什么区别： 第二行编译得很好，但我的 Ecplipse IDE 抱怨“声纳：用方法引用替换这个 lambda”。 如何选择要使用的适当代码？这些都有具体的用例吗？

我将为消息实现断路器模式。基本要求是，如果微服务无法将消息发布到发布主题，则应停止接受来自其他 Kafka 主题的消息。当发布主题可用时，微服务应开始接受来自其他 Kafka 主题的消息。 有没有一种方法可以在Spring BootKafka Streams中实现这一点？

我正在使用 Spring靴：2.3.5.释放Spring云：Hoxton.SR8 我正在尝试Spring云流Kafka流应用程序。一切都运行良好，直到出现反序列化异常。应用程序每次都会关闭。 我想跳过不良记录，在Kafka主题中前进。但我无法实现这一点。配置： 我得到的错误是 现在我正在使用这个设置。它仍然没有效果。根据留档，它应该简单地记录错误并继续处理。即它应该跳过不良记录。但这并没有发生。看

我编写了一个Spring Cloud Streams Kafka Streams Binder应用程序，它将多个Kafka输入主题多路复用到一个流中: (来源:https://spring . io/blog/2019/12/03/stream-processing-with-spring-cloud-stream-and-Apache-Kafka-streams-part-2-programmi

我们正在使用Avro输入/输出记录通过Spring Cloud Stream功能支持测试Kafka Streams的使用，但设置和，以便在执行Avro转换的地方使用自定义的。 默认的 和值的 。 当我们只使用KStream到KStream函数时，一切都没问题，例如: 但是当我们尝试一个稍微复杂一点的例子，涉及一个像这样的输入KTable: （ 类只有两个成员：） 收到第一条记录时，将引发此异常：

我已经使用Spring Cloud Stream和Spring Cloud函数创建了一个Kafka消费者，用于以批处理模式从Kafka主题消费消息。现在，我想将错误批处理发送到死信队列以进一步调试错误。 我正在使用Spring重试处理消费者方法中的重试。但对于不可重试的异常，我希望将整个批次发送到DLQ。 这是我的消费者的样子： 错误处理配置如下所示: 使用DeadRec总计PublishinRe

我有一个场景，比如。 有4个主题处理主题，异常主题，重试主题和拒绝主题。我有一个spring cloud stream应用程序，它有一个使用Kstream的处理器。该处理器从异常主题中读取消息，并基于每个消息中可用的标志，为重试主题和拒绝主题创建两个kstream分支。现在需要做的是，retry主题中出现的任何消息都必须等待一段特定的时间，然后才能将其推回处理主题。有没有人能帮我在spring c

我们刚刚升级到Spring-Cloud-Stream的3 . 0 . 0-版本，遇到了以下问题: 像这样使用函数式风格时: application.yaml如下所示： 似乎序列化做了两次——当我们截取kafka主题中产生的消息时，消费者只是将它们显示为JSON(字符串),但现在它是一个不可读的字节[]。此外，生产中的下游消费者不能再反序列化消息。奇怪的是，输入消息的反序列化似乎工作得很好，无论我们

以下是我的情况： 我们有一个Spring cloud Stream 3 Kafka服务连接到同一个代理中的多个主题，但我想基于属性控制连接到特定主题。 每个主题都有自己的活页夹和绑定，但代理对所有人来说都是一样的。 我尝试使用下面的属性禁用绑定（这是我到目前为止找到的唯一解决方案），这适用于StreamListener不接收消息，但与主题的连接和重新平衡仍在发生。 我想知道活页夹级别是否有任何设置

我有来自 3 个 mysql 表、1 个主表和两个子表的原始流。我尝试加入三个原始流并转换为单个输出流。如果父流上有任何更新，但如果子流发生任何变化，则不触发输出，它就可以工作。 父流上的任何新添加或更新都由处理器拾取，并将其与其他KTable连接，并在输出流上返回。但对child1stream或child2stream的任何添加或更新都不会触发输出流。 我认为将所有输入流设为 KTable，它们

我有一个处理器，它从主题中获取json字符串，类型为GenericRecord。现在我把这条河分成两条支流。我采用第一个分支，并将（key，value）映射为2个字符串，其中包含一个特定的json字段和该字段的值，然后按key分组。到目前为止，一切都很好。现在，我必须用用户定义的新类型聚合流，并收到一个异常。 这里是代码： 新类型： 好流： 问题是: 这是例外： 我如何解决这个问题？ 更新 ---