Spring Cloud Stream -查询主题而不显式消费KTable/KStream？
Spring Cloud Stream是一个用于构建消息驱动微服务的框架。它提供了一种简化和标准化的方式来进行事件驱动的通信，同时隐藏了底层消息传递系统的复杂性。通过Spring Cloud Stream，开发人员可以轻松地使用消息传递来连接和集成不同的系统组件。
在Spring Cloud Stream中，查询主题而不显式消费KTable/KStream是指通过订阅一个消息主题，并且从中获取消息的内容，而不需要直接处理或消费消息中的KTable或KStream。这种方式适用于需要获取消息主题中的数据内容，但不需要对数据进行处理或转换的场景。
为了实现查询主题而不显式消费KTable/KStream，可以使用Spring Cloud Stream提供的绑定器和消息转换器。绑定器负责连接消息传递系统，而消息转换器用于将原始消息转换为Java对象。
以下是使用Spring Cloud Stream查询主题而不显式消费KTable/KStream的步骤：
- 创建一个Spring Boot应用程序，并添加Spring Cloud Stream和所需的绑定器依赖。
- 在应用程序中定义一个接口，使用@Input注解标记输入通道，并指定要订阅的消息主题。例如：
- 在应用程序中定义一个接口，使用@Input注解标记输入通道，并指定要订阅的消息主题。例如：
- 在应用程序中创建一个消息处理器，用于处理从主题接收到的消息。在处理器方法上使用@StreamListener注解，并指定要使用的输入通道。例如：
- 在应用程序中创建一个消息处理器，用于处理从主题接收到的消息。在处理器方法上使用@StreamListener注解，并指定要使用的输入通道。例如：
- 在应用程序的配置文件中配置绑定器和消息转换器的相关属性。例如，可以指定要使用的消息传递系统和消息序列化方式。
- 在应用程序的配置文件中配置绑定器和消息转换器的相关属性。例如，可以指定要使用的消息传递系统和消息序列化方式。
通过上述步骤，应用程序将能够查询主题并获取消息内容，而不需要显式消费或处理KTable/KStream。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CMQ：腾讯云提供的高可用分布式消息队列服务，支持高并发消息传递和可靠性保证。详情请参考：CMQ产品介绍
- 腾讯云消息队列 CKafka：腾讯云提供的Apache Kafka兼容的分布式消息队列服务，适用于大规模流式数据处理和实时分析场景。详情请参考：CKafka产品介绍
相关·内容
我在一个Java应用程序中使用了Spring Cloud Stream库。我想为状态存储使用Kafka Streams绑定器。应用程序将向主题发布消息，我希望使用Kafka Streams InteractiveQueryService来检索同一主题中的数据。是否可以按原样执行这样的查询，或者我是否需要首先将主题作为KTable/KStream使用并物化它，然后才能执行查询？我不需要对topic进行KTable/KStream处理，我只想查询topic的内容。我希望有某种方法可以隐式地将其具体化为状态存储。
浏览 36提问于2020-01-28得票数 0
我正在努力寻找任何关于我可以在哪里使用Spring Cloud Streams的文档，这些文档将Kafka主题放入KTable中。在使用注解在Spring boot中执行此操作的过程中，我已经查找了文档，例如这里的https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/3.0.0.RC1/reference/html/spring-cloud-stream-binder-kafka.html#_materializing_ktable_as_a_state_store，但没有什么具体的内容。我希
浏览 19提问于2021-04-22得票数 0
我使用编写了一个Kafka应用程序，无法在KTable中看到任何消息。我无法追踪这个问题。任何指点或帮助都值得赞赏。
下面的是代码
import java.util.function.Function;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
imp
浏览 1提问于2019-11-16得票数 1
回答已采纳
1回答
我正在写一个Spring Boot (2.1.4)应用程序，试图使用Spring Cloud Streams for Kafka。 我要做的是维护一个主题的传感器列表(“传感器”)。OTOH，我有另一个主题的传入数据(“数据”)。我试图实现的是，当我获得一个传感器的数据时，我希望将其添加到传感器列表中。 为此，我从传感器主题创建了一个KTable<String, Sensor>，将温度主题映射到纯传感器的数据(在本例中是它的名称)，并与一个ValueJoiner进行外部连接，如果传感器存在，则保留传感器，否则使用读数的传感器。然后，我将结果写回sensors主题。 KTable&
浏览 29提问于2019-05-03得票数 1
KTable<Key1, GenericRecord> primaryTable = createKTable(key1, kstream, statestore-name);
KTable<Key2, GenericRecord> childTable1 = createKTable(key1, kstream, statestore-name);
KTable<Key3, GenericRecord> childTable2 = createKTable(key1, kstream, statestore-name);
primaryTable.le
浏览 12提问于2019-01-27得票数 0
回答已采纳
1回答
我正在尝试构建并通过KTable通道获得SCSt。但这是行不通的。输入KTable没有数据，但是如果我试图查看KSTream聚合(toStream())，我可以看到一些数据。我明白了，KTable不是queryable，也没有queryable名称。
类：
@Slf4j
@EnableBinding({LimitBinding.class})
public class CommonWorker {
@Value("${app.dataflow.out-destination}")
private String customerOut;
private Limit
浏览 1提问于2018-07-09得票数 0
回答已采纳
1回答
如何使用使用主题中的最新记录？
、、、、
我有一个Kafka流处理器，它消耗三个主题，并试图在键上合并(加入操作)它们。成功加入后，它进行一些聚合，然后将结果推送到目标主题。应用程序第一次运行后，它将尝试使用来自这些主题的所有数据。其中两个主题使用类似查找表，这意味着我需要从一开始就使用所有数据。但其中一个主题是我的主题。所以我需要用最新的。但我的应用程序从一开始就消耗了卡夫卡的所有主题。因此，我想从一开始就学习两个主题和一个最新的主题。我正在使用，。这是我的心声和一些代码片段；
Application.yaml：
spring.cloud.stream.function.definition: processName;
spring
浏览 10提问于2022-02-05得票数 0
回答已采纳
我们有一个场景，我们希望在cluster#1上消费来自kafka主题的数据，但在cluster#2上创建KTable主题(重新分区和changelog)。
通道绑定-
spring.cloud.stream.bindings.member.destination: member
spring.cloud.stream.bindings.member.consumer.useNativeDecoding: true
spring.cloud.stream.bindings.member.consumer.headerMode: raw
spring.cloud.stream.kafka.stre
浏览 4提问于2019-01-12得票数 0
我正在尝试使用输入主题作为KTable对记录进行重复数据删除，并将它们下沉到输出主题。但KTable仍在将重复记录下沉到输出主题。不确定我哪里错了。 这是我的application.yml spring:
cloud:
stream:
function:
bindings:
process-in-0: input.topic
process-out-0: output.topic
definition: process
kafka:
streams:
浏览 38提问于2021-05-07得票数 1
回答已采纳
参见下面的更新以显示潜在的解决办法
我们的应用程序使用两个主题作为KTables，执行一个左连接，并输出到一个主题。在测试期间，我们发现当我们的输出主题只有一个分区时，它可以正常工作。当我们增加分区的数量时，我们注意到生成到输出主题的消息的数量减少了。
在启动应用程序之前，我们用多个分区配置测试了这个理论。使用一个分区，我们可以看到100%的消息。对于2，我们看到一些消息(少于50%)。只有10，我们几乎没有看到(不到10%)。
因为我们要加入，所以从主题1中消耗的每一条消息都应该被写入输出主题，但是我们发现这种情况并没有发生。消息似乎被卡在从Ktable的外键连接创建的“中间”主题中，但是没
浏览 0提问于2020-07-13得票数 8
回答已采纳
我有一个基于Spring的绑定应用程序。它定义了一个包含以下部分的拓扑：
绿色中的数字显示了通过通过流绑定绑定的各个处理器定义的拓扑传递的消息数量，以下是各自的属性：
spring.cloud.stream.bindings:
...
hint1Stream-out-0:
destination: hints
realityStream-out-0:
destination: hints
countStream-in-0:
destination: hints
我正在计算每个处理器使用peek()方法生成/使用的消息如下：
return str
浏览 1提问于2020-05-30得票数 2
回答已采纳
我们使用2.1.3版本的spring cloud stream kafka streams - <dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
<version>2.1.3.RELEASE</version>
</dependency&
浏览 12提问于2019-10-11得票数 1
我正在使用Spring Boot Apache Kafka示例，并得到下面的错误。
Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'ÿ': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
at [Source: (byte[])"�contentType
at com.fasterxml.ja
浏览 0提问于2021-03-05得票数 1
我正在做一个POC来演示一些Spring Cloud Kafka功能。我使用的是带有SpringBoot 2.4.1的Java11。我的build.gradle具有以下库
implementation 'org.apache.kafka:kafka-streams'
implementation 'org.springframework.cloud:spring-cloud-stream'
implementation 'org.springframework.cloud:spring-cloud-stream-binder-ka
浏览 0提问于2021-01-30得票数 0
我是Spring Cloud和kafka stream的新手。我正在尝试使用kafka活页夹设置spring cloud应用程序。我尝试在本地测试kafka流处理器，但无法打印任何日志。 我的kafka消息将包含JSONObject。kafkaStreamListener类是： @Configuration
public class KafkaStreamListener {
private static Logger logger = LogManager.getLogger(KafkaStreamListener.class);
//bean for proc
浏览 37提问于2021-07-27得票数 0
回答已采纳
我有一个复杂的Kafka应用程序，其中两个流在同一个流中完全有状态：
它使用Execution主题作为源，增强消息并将其重新发布回相同的Execution主题。it加入另一个主题WorkerTaskResult，将结果添加到Execution并发布回Execution主题.
主要目标是提供一个工作流系统。
去尾逻辑是：
the an 是一个 Execution列表，查看所有TaskRun的所有当前状态，找到下一个要执行(如果有)的列表，执行更改其TaskRunsList并添加下一个TaskRunsList并发布回卡夫卡，另外，它向另一个队列发送要执行的任务，(WorkerTask)the Wo
浏览 5提问于2020-04-20得票数 1
回答已采纳
我有一个应用程序，它从一个主题多分区读取kstream，而从另一个主题多分区读取ktable。因此Kstream和Ktable都将引用两个不同主题中相同分区或两个不同主题中的任何分区
浏览 13提问于2021-10-26得票数 0
1回答
我有一个主题名叫push-processing-KSTREAM-PEEK-0000000014-repartition，这是卡夫卡的内部话题。我没有创建这个主题，我在重新分区后使用.peek()方法，并使用peek方法3-4次。
我的问题是，我可以阅读主题topic read push-processing-KSTREAM-PEEK-0000000014-repartition，但我不能阅读，当我说topic read push-processing-KSTREAM-PEEK-0000000014-repartition --from-beginning。
创建这个内部主题是因为使用了peek
浏览 2提问于2019-11-22得票数 0
我正在尝试加入一个KStream和一个KTable。如果没有联接，我从中间主题"book-attribute-by-id“中阅读就没有问题。
用于KTable的示例msg：
{key: {id: 1}
value: {id: 1, attribute_name: "weight"}}
用于KStream的示例msg：
{key: {id: 1},
value: {id: 1, book_id: 1, attribute_id: 1, value: 200}}
希望输出到“最终聚合”主题：
{key: {id: 1},
value: {book_id: 1, at
浏览 1提问于2019-02-06得票数 0
回答已采纳
当使用KTable时，当实例/使用者的数量等于分区数时，Kafka流不允许从特定主题的多个分区读取实例。我试着用GlobalKTable实现这一点，问题是数据将被覆盖，聚合也不能应用于它。
假设我有一个名为"data_in“的主题，它有3个分区(P1、P2、P3)。当我运行一个Kafka流应用程序的3个实例(I1、I2、I3)时，我希望每个实例从"data_in“的所有分区读取数据。我的意思是，I1可以从P1、P2和P3中读取，I2可以从P1、P2和P3、I2和on中读取。
编辑:请记住，生产者可以在"data_in“中将两个类似的ID发布到两个不同的分区中。因此，当运
浏览 0提问于2018-12-11得票数 1
回答已采纳
我正在使用Spring Cloud Stream处理Kafka Streams。在消息处理应用程序中，可能会产生错误。因此，不应提交并重试该消息。
我的申请方法-
@Bean
public Function<KStream<Object, String>, KStream<String, Long>> process() {
return (input) -> {
KStream<Object, String> kt = input.flatMapValues(v -> Arrays.asList(v.toUpperCase().sp
浏览 16提问于2020-06-04得票数 1
回答已采纳
是否可以在Spring Cloud Stream中使用带有@EnableBinding注释的类或带有@StreamListener的方法中的交互式查询(InteractiveQueryService)？我尝试在提供的KStreamMusicSampleApplication类和进程方法中实例化ReadOnlyKeyValueStore，但总是为空。 我的@StreamListener方法正在监听一堆KTables和KStreams，在处理拓扑的过程中，例如过滤，我必须检查来自KStream的键是否已经存在于特定的KTable中。 我试图弄清楚如何扫描传入的KTable以检查密钥是否已经存在，
浏览 16提问于2019-05-01得票数 1
回答已采纳
我正在尝试从一堆数据点创建一个窗口存储，但由于某种原因，似乎没有对流拓扑分支进行评估。 我使用相同的流来填充实例化到存储中的KTable，这很好用。 我使用Spring Cloud Streams，配置如下： spring:
application.name: stream-test
kafka.bootstrap-servers: localhost:9092
cloud.stream:
# assign group and topic name to binding
bindings:
windowedStream:
desti
浏览 19提问于2019-05-09得票数 1
回答已采纳
1回答
这是我的应用程序，它简单地从客户主题(输入绑定)获取对KStream的引用，从order主题获取另一个引用(订单绑定)。然后，它从客户主题创建一个KTable，并与order KStream执行连接：
@Configuration
class ShippingKStreamConfiguration {
@StreamListener
@SendTo("output")
fun process(@Input("input") input: KStream<Int, Customer>, @Input("order&
浏览 1提问于2019-04-17得票数 0
回答已采纳
我正在尝试更好地理解如何设置集群来运行我的Kafka-Stream应用程序。我正在尝试对将要涉及的数据量有更好的认识。
在这方面，虽然我可以很快看出KTable需要状态存储，但我想知道从主题创建Kstream是否意味着将该主题的所有日志复制到状态存储中，显然是以一种仅附加的方式。也就是说，尤其是当我们想要公开流以进行查询时？
当数据在源主题中移动时，Kafka会自动复制状态存储中的数据吗？如上所述，由于Ktable的更新，这听起来很明显，但对于Kstream，我只想确认发生了什么。
浏览 17提问于2019-07-09得票数 1
回答已采纳
我有一个Kafka主题，我在其中发送定位事件(key=user_id，value=user_location)。我能够将其作为KStream读取和处理
KStreamBuilder builder = new KStreamBuilder();
KStream<String, Location> locations = builder
.stream("location_topic")
.map((k, v) -> {
// some processing here, omitted form clar
浏览 0提问于2017-03-22得票数 38
回答已采纳
当使用Kafka流DSL时，是否有一种方法可以使用相同的主题作为两个不同处理例程的源？
StreamsBuilder streamsBuilder = new StreamsBuilder();
// use the topic as a stream
streamsBuilder.stream("topic")...
// use the same topic as a source for KTable
streamsBuilder.table("topic")...
return streamsBuilder.build();
上面的简单实现在运行时
浏览 0提问于2018-09-20得票数 5
回答已采纳
1回答
我引用了发布在上的示例。我正在尝试同时运行多个spring cloud stream应用程序。这里，first的输出作为其他的输入。下面是我正在尝试做的事情。
@Bean
public Function<KStream<FormUUID, FormData>, KStream<UUID, Application>> process()
{
//do some processing here and return
}
// read output from above process and join it with
浏览 10提问于2020-05-19得票数 1
回答已采纳
1回答
我有一个场景，在这个场景中，我需要在Kafka中导入一个完整的DB，并在DB术语中创建一些用户可以查询的表的视图。我的要求是通过物理模型(表)中的视图重新构建逻辑模型。
因此，我想知道这样做的步骤。
我的理想是，kafka创建与表相对应的主题，然后由我声明性地(使用KSQL)创建视图。
虽然我在这里描述的东西一开始听起来是可行的，但是我对数据--主题中的数据的结构(模式)--有一个问题。问题似乎是，我可能不得不做一个额外的步骤，但不知道它是否可以避免或实际上是必要的。
更具体地说，视图通常表示表上的联接。我想，如果我想要在表上加入，我需要已经创建了Ktable或Kstream，这给出了要在其上
浏览 2提问于2019-07-06得票数 0
回答已采纳
我正在学习apache Kafka网站link的教程。 输入主题被处理为流，中间主题也被生成，但最终输出主题为空。 以下是拓扑输出： Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-0000000000 (topics: [input])
--> KSTREAM-FLATMAPVALUES-0000000001
Processor: KSTREAM-FLATMAPVALUES-0000000001 (stores: [])
--> KSTREAM-KEY-SELECT-0000000002
<-- KSTRE
浏览 40提问于2020-11-02得票数 0
回答已采纳
我用春云条纹Kstream。我测试一个主题&一个@StreamListner。没事的。
我为两个KStream输入修改我的代码。(两个@StreamListener)但是，弹簧云错误..。
***************************
APPLICATION FAILED TO START
***************************
Description:
The bean 'stream-builder-process', defined in null, could not be registered. A bean with that
浏览 0提问于2019-03-13得票数 4
回答已采纳
我们正在尝试实现下面描述的用例，我们有我们希望克服的实现问题，
用例，
我们正在尝试通过匹配两个流的消息中存在的关键字(JSON)来在两个Kafka主题之间进行KStream连接。此外，我们应该维护消息序列，因为它是从源文件到达KStream的。
场景是，如果匹配的键还没有到达任何一个流，我们应该停止或重试join，直到预期的键到达其他主题。我们想把不匹配的记录放回KStream，但在这种情况下，不能保证顺序。
问题1:如何在其他主题中停止或保持join，直到预期的键到达。例如，KTable有密钥100，但KStream还没有收到密钥100，那么我们应该重试Join或按住KStream，直到密
浏览 2提问于2019-05-08得票数 2
是否有可能从使用者那里创建一个物化视图，而无需真正为其编写逻辑？因为我在kafka主题中的数据可以在没有任何转换的情况下实现。
1. Using KTable
@Bean
public Consumer<KTable<String, Pojo>> process() {
return ktable -> <what should i do here>??;
}
我在文档中看到，可以直接执行spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.materialized
浏览 1提问于2021-07-26得票数 0
回答已采纳
1回答
我知道Kafka允许基于指定谓词将数据分发到多个主题，而且Kafka绑定器使用@StreamListener和functional binding方法都支持这一点。
...
// return type KStream<?, WordCount>[]
Predicate<Object, WordCount> isEnglish = (k, v) -> v.word.equals("english");
Predicate<Object, WordCount> isFrench = (k, v) -> v.word.equals
浏览 1提问于2021-04-02得票数 1
回答已采纳
我正在尝试使用Spring Cloud Stream创建一个简单的功能bean，它处理来自KStream和GlobalKTable的消息，连接它们，聚合它们，并将结果输出到一个新的流中，但我在正确配置它所需的serde方面遇到了困难。
无需多说，下面是我的方法：
@Bean
public BiFunction<KStream<GenericRecord, GenericRecord>, GlobalKTable<Long, GenericRecord>, KStream<String, MyCustomJavaClass>> joinAndAgg
浏览 82提问于2020-05-20得票数 1
回答已采纳
我有一个主题，在我的流处理器中有多个分区，我只想从一个分区中流式传输它，并且可以弄清楚如何配置它
spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=s-processor
spring.cloud.stream.bindings.input.destination=uinput
spring.cloud.stream.bindings.input.group=r-processor
spring.cloud.stream.bindings.input.contentType=application/
浏览 1提问于2019-02-16得票数 0
我正在使用Spring Cloud Streams，我有一个接收KStream和GlobalKTable的BiFunction。我不想在应用程序重新启动后丢失GlobalKTable数据，但这正是正在发生的事情。
@Bean
public BiFunction<KStream<String, MyClass1>, GlobalKTable<String, MyClass2>, KStream<String, MyClass3>> process() {
...
}
我还配置了"materializedAs“属性：
spring.clou
浏览 3提问于2021-05-14得票数 0
2回答
即使在特定配置之后，Spring流也不会在反序列化错误上重试。期望是，它应该基于配置的重试策略重试，最后将失败的消息推送到DLQ。
配置如下。
spring.cloud.stream.bindings.input_topic.consumer.maxAttempts=7
spring.cloud.stream.bindings.input_topic.consumer.backOffInitialInterval=500
spring.cloud.stream.bindings.input_topic.consumer.backOffMultiplier=10.0
spring.cloud.s
浏览 1提问于2019-06-26得票数 0
回答已采纳
KStream<String, String> kstream = builder.stream("input-topic");
kstream.to("output-topic");
已经创建了“输入-主题”。我没有创建“输出主题”，似乎"Kstream“和其他内部主题一起为我创建了一个主题。此外，在"to“函数的javadoc中看到了这一点，指定的主题应该在使用之前手动创建(即在Kafka流应用程序启动之前。
所以我的问题是，我们总是必须手动创建“输出主题”吗？
浏览 0提问于2021-10-16得票数 0
回答已采纳
有没有使用Spring Cloud stream和函数式方法创建GlobalKTable以从KStream中记录计数的示例？
浏览 4提问于2020-03-25得票数 0
我是卡夫卡的新手，我想要帮助
我有两个应用程序(springboot)，它们是相同的/拷贝只有不同的端口。
他们都是消费者
这两个人听XXX这个话题
我有另一个应用程序扮演制片人的角色。
每当我给某某话题发东西。
只有其中一个使用消息，而另一个不使用消息。
我分别测试了两种方法，如果他们是单独听的话，他们也会正常地听，但是如果他们在一起听，只有一人会听。
我在用
<dependency>
<groupId>org.springframework.cloud</groupId>
<a
浏览 0提问于2021-07-06得票数 1
回答已采纳
我有一个主题，它接收带有以下信息的事件：
key -> orderId (整数)
value -> {"orderId" : aaa, "productId" : xxx, "userId" : yyy, "state" : "zzz"} (包含订单完整信息的JSON)
我想通过orderId实现一个交互式查询来获得完整的订单信息。这个想法是能够从物化视图( Kafka Streams存储)获得订单的当前状态。
首先，我创建主题的KStream：
KStream<Integer, JsonNode
浏览 8提问于2017-07-17得票数 1
回答已采纳
我正在做一个关于Kafka流和ktables的poc，我想知道是否有任何方法可以在kafka中存储数据(键-值对或键-对象对)，可以通过流、ktables、状态存储，这样我就可以根据键和值检索数据库。我创建了一个基于主题的kstream，在这个kstream上我推送了一些消息，并使用wordcountalgo在kstream上面创建的ktable中填充值。如下所示：
StoreBuilder customerStateStore = Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("customer-store
浏览 0提问于2019-07-24得票数 2
我有一个基于Webflux的微服务，它有一个简单的反应式存储库： public interface NotificationRepository extends ReactiveMongoRepository<Notification, ObjectId> {
} 现在我想扩展这个微服务来消费来自Kafka的事件消息。然后，此消息/事件将保存到数据库中。 对于Kafka监听器，我使用了Spring Cloud Stream。我创建了一些简单的Consumer，它工作得很好-我能够消费消息并将其保存到数据库中。 @Bean
public Consum
浏览 40提问于2020-12-15得票数 4
回答已采纳
在我的制作人application.properties中，我有
spring.cloud.stream.bindings.output.producer.partitionKeyExpression=payload
spring.cloud.stream.bindings.output.producer.partitionCount=4
我的消费者application.properties看起来像这样
spring.cloud.stream.bindings.input.consumer.partitioned=true
spring.cloud.stream.bindings.input
浏览 0提问于2016-08-18得票数 0
我在我的应用程序中实现了一个KTable和一个KStream连接，并在以下情况下期望输出消息；
：KStream中的新消息，KTable 中有匹配记录，KStream中有更新的消息，当KTable记录上有更新时，KTable中有匹配记录
我已经观察到我的应用程序可以像预期的那样执行1和2，而不是3。
有什么建议可以让我达到第三点吗？
谢谢!
浏览 3提问于2021-02-10得票数 0
需要在Spring绑定器配置中使用KTable，下面是用于从主题读取数据并在控制台上打印数据的示例代码。
但是，在应用程序启动时，以抛出java.lang.IllegalArgumentException:方法结束的应用程序必须是声明式。
转介，结果仍然一样
爪哇- 11
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2
浏览 0提问于2021-03-19得票数 0
2回答
我有一个具有函数编程风格的Spring流的工作设置。有两个用例，它们是通过application.properties配置的。它们各自工作，但当我同时激活这两者时，就会得到第二个用例的输出流的序列化错误：
Exception in thread "ActivitiesAppId-05296224-5ea1-412a-aee4-1165870b5c75-StreamThread-1" org.apache.kafka.streams.errors.StreamsException:
Error encountered sending record to topic outputA
浏览 2提问于2020-11-25得票数 1
回答已采纳
1回答
是否可以使用两个函数，其中第一个函数的输出是第二个函数的输入？
我的职责：
@Configuration
public class StringStream {
@Bean
public Supplier<Flux<String>> stringSupplier() {
return () -> Flux.interval(Duration.of(1L, ChronoUnit.SECONDS))
.map(v -> RandomStringUtils.randomAlphabetic(10))
浏览 2提问于2021-01-06得票数 0
我正在尝试编写我的第一个练习应用程序，使用Kafka Streams来计算主题中的字数。但是，我想我指的是旧的API，因为在lambda函数的末尾，我想将KTable的输出放到一个主题中，但我没有看到任何这样的方法。 我引用的代码使用了to()方法，但我认为现在没有这样的方法。我看到了toStream()，但不知道如何使用它将消息放到特定的输出主题中。 有没有人可以看一下，因为这应该是非常基础的。 public static void main(String[] args) {
Properties config = new Properties();
co
浏览 19提问于2019-12-26得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
