《stream》专题

我正在做一个项目，使用springboot、Spring cloud Netflix等来构建微服务。 对于一些异步通信，我使用Spring Cloud Stream来生成和使用事件。e、 g.在合同服务中起草业务合同后，该服务发布一个合同创建事件，该事件将由审计服务使用，以初始化审计过程。此外，用户服务将使用该事件为相关方创建通知。 场景是我有很多事件，消费者会根据事件类型订阅感兴趣的事件。我的问

我把keyby parallism设置为30，我如何找到30个唯一的键来使datastream键控均匀？flink支持这个吗？因为我不想要太多的键，我会在influxdb中把键作为一个标记，所以我不想要太多的键，避免在influxdb中使用oom。但是我如何使用最小键来均匀地按一个datastream键呢？ 我想使用flink来跟踪MySQL中表的每一次更改（如UPDATE/INSERT）,并计算

假设我有一个字符串流： 工作正常： 比较器是一种功能接口。这意味着我们可以使用方法引用或lambdas来实现它。Comparator接口实现一个方法，该方法接受两个字符串参数并返回一个int。但是，Comparator：：ReverseOrder不这样做。它是对接受零参数并返回比较器的函数的引用。这与接口不兼容。这意味着我们必须使用方法，而不是方法引用。 但我不明白。

我正在尝试使用Spring Cloud Stream框架构建一个简单的Kafka Streams应用程序。我可以连接到流以推送原始数据进行处理。但是当我尝试按键处理流进行事件计数时，我得到了未找到的运行应用程序时异常。我检查了我的项目包含的库，我可以找到类，它没有丢失。我不确定为什么在运行时它没有被加载！ 下面是我的源文件。 <代码>com。pgp。学Kafka。分析。分析应用程序 <代码>com

我正在使用spring kafka来实现一个使用spring Boot 1.5.16的流应用程序。我们使用的SpringKafka版本是1.3.8。释放。 我正在搜索一种方法，以便在出现终止与Kafka流关联的所有线程的错误时关闭启动应用程序。我发现在KafkaStreams中，有可能为未捕获的异常注册句柄。方法是setGlobalStateRestoreListener。 我看到这个方法在类型k

参考本文档的4.2.6https://docs.spring.io/spring-kafka/reference/htmlsingle/#kafka-streams 如何使用kafka stream spring支持访问州立商店？ 没有Spring你还能做什么？ 但我不知道如何才能接触到Kafka斯特雷姆的目标。

我对弹性APM非常陌生，不知道它如何支持不同的框架。我可以从文档中看到，APM支持Spring Boot。我已经用APM测试了一个Spring Boot应用程序，它看起来很有希望。我想知道APM是否也支持Spring Cloud Stream。Spring Cloud Stream通过使用Spring Boot和消息传递中间件提供事件驱动的体系结构。中间件可以是Kafka、RabbitMQ等。

澄清：请注意，这个问题与这个问题不同：如何使用Spring Cloud Stream Kafka和每个服务的数据库实现微服务事件驱动架构 这一个是关于使用Kafka作为唯一的存储库（事件），不需要DB，另一个是关于使用每个服务的数据库（MariaDB）。 我想实现一个事件源架构来处理分布式事务： OrdersService接收订单请求并将新订单存储在代理中。 这是我的主要疑问：我如何查询Kafka

Pom文件包含依赖项-spring cloud starter流kafka 控制器代码 问候听众类 问候语流接口 问候服务 问候课程 结合 主类 我能够使用属性文件中指定的以下配置连接、发送和接收消息到本地Kafka实例 但是，我无法连接到blue mix云上的IBM事件流。下面是我连接到云上事件流的配置 请让我知道配置有什么问题。我没有找到任何符合我要求的例子。

我正在开发Camunda BPM Spring Boot应用程序。该应用程序使用Spring Cloud Stream从Rabbitmq队列中读取消息。一旦收到消息，应用程序就会调用Camunda中的进程实例。 如果在应用程序启动期间rabbitmq队列中已经存在消息，则云流侦听器甚至在初始化Camunda之前就开始读取消息。 是否可以停止云流侦听器侦听队列，直到触发某个事件为止？在本例中为Pos

我正在尝试使用Spring Cloud Stream来发布和使用Kafka消息。我一直在访问绑定通道上处理留档。我正在尝试为我的主题在通道上使用自定义名称，因此当我尝试注入它时，我有一个@Qualifier，但Spring找不到相关的bean。它说“对于每个绑定接口，Spring Cloud Stream将生成一个实现该接口的bean”，但自动连接不起作用。 我得到的错误是“com中构造函数的参数

我正在使用带有kafka绑定的spring cloud streams。消费者使用spring进行配置。云流动Kafka。绑定。功能配置和DLQ已启用。 当我使用来自主题的消息并发生异常时，我看到它重试3次，并将消息发送到DLQ。我已将maxAttemts配置为5，但我无法覆盖默认值3。 我使用的是spring kafka（2.7.8）和spring cloud（2020.0.4）。如何覆盖重试尝

我正在使用Kafka活页夹的Spring Cloud Stream。它工作得很好，但客户端接收到重复的消息。已经尝试了所有Kafka消费属性，但没有结果。 在我的应用程序示例中检查2个类-Aggregate Application和EventFilterApplication。如果我运行EventFilterApplication-只有1条消息，如果是Aggregate Application-2

尝试将Spring配置为在使用批处理模式时向死信队列发送错误消息。但由于dlq主题中没有任何内容。 我使用Spring Boot 2.5.3和Spring Cloud 2020.0.3。这会自动将spring cloud stream binder kafka父版本解析为3.1.3。 这是申请表。属性： 这是函数式编程模型中的应用程序和批处理侦听器： 发送到主题： 从DLQ阅读： 因此，运行此应用

我有一个使用Spring Cloud Stream（v1.3.0）和Kafka（v1.1.6）的Spring boot（v.1.57）应用程序。我希望能够优雅地关闭它，即关闭时，所有流监听器（即用@StreamListener注释）应该： 停止轮询新消息 完成他们的工作 将偏移提交给Kafka 我注意到ContainerProperties（默认设置为10000ms）中有一个名为“Shutdown