对多个Kafka主题重用一个模式
是指在使用Apache Kafka消息队列系统时，可以将同一个消息模式（Schema）应用于多个主题（Topic）。消息模式定义了消息的结构和字段，使得生产者和消费者能够理解和解析消息内容。
优势：
- 代码复用：通过重用一个模式，可以减少开发人员编写和维护的代码量，提高开发效率。
- 数据一致性：使用相同的模式，可以确保不同主题中的消息具有一致的结构，避免数据解析错误和混乱。
- 简化管理：减少了维护多个模式的工作量，简化了模式的管理和更新过程。
应用场景：
- 多主题数据同步：当多个主题包含相似或相关的数据时，可以使用相同的模式来确保数据一致性，方便进行数据同步和分析。
- 多个消费者组：当存在多个消费者组需要消费相同结构的消息时，可以重用一个模式，避免重复定义和管理多个模式。
推荐的腾讯云相关产品： 腾讯云提供了一系列与消息队列相关的产品，其中包括云原生消息队列 CMQ、云消息队列 CKafka 等。这些产品可以满足不同规模和需求的消息队列场景。
- 腾讯云云原生消息队列 CMQ：是一种高可靠、高可用的分布式消息队列服务，适用于异步通信、流量削峰、解耦、日志处理等场景。详情请参考：云原生消息队列 CMQ
- 腾讯云云消息队列 CKafka：是一种高吞吐、低延迟的分布式消息队列服务，适用于大数据实时计算、日志采集、流式处理等场景。详情请参考：云消息队列 CKafka
相关·内容
2回答
我正在看如何记录所有卡夫卡主题/事件。有什么env变量可以运行我的Kafka容器吗？为了开发目的，我用了
docker run -p 9092:9092 -e ADVERTISED_HOST=127.0.0.1 johnnypark/kafka-zookeeper
当我运行python脚本时，我能够使用和发布事件。
from kafka import KafkaConsumer
consumer = KafkaConsumer('other-topic')
for msg in consumer:
print (msg)
from kafka import Kafk
浏览 9提问于2021-06-18得票数 1
回答已采纳
2回答
目前，我有几个运行良好的集成测试，但是我正在使用注释@QuarkusTestResource在夸克启动之前通过测试容器启动容器，并通过重写扩展QuarkusTestResourceLifecycleManager的容器类的开始方法来调整属性以适应随机测试容器端口。
为了进行一些优化，例如，我希望重用启动kafka的容器，因为启动它需要6-8秒，并在我的几个集成测试中重用它。我一直没能这么做。因为Quarkus管理生命周期，每次它停止，在每个测试类之间，它也停止每个容器。我尝试将与Quarkus测试资源混合使用，但它不起作用。下面是我的一个集成测试类开始的片段：
@QuarkusTest
@Ta
浏览 9提问于2022-03-30得票数 2
回答已采纳
我启动了kafka连接镜像。我配置了近25个正在运行的源和宿连接器。当我进入这个容器时，只看到了1个java进程。
root@connect:/# ps -ef | grep java
root 1 0 3 Jun20 ? 01:32:06 java -Xms256M -Xmx2G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=tr
浏览 1提问于2018-06-22得票数 1
我制作了一个坞-撰写文件来设置Kafka和模式注册表。但是，我无法检查模式注册表集群是否运行良好。
如何检查架构注册表群集模式？
我的坞-撰写文件作为集群运行良好吗？我不确定它的效果是否好。我只能看到两个卡夫卡和两个注册表在运行。
这是我的拼图文件。
version: '2'
services:
zookeeper:
image: zookeeper:3.7.0
container_name: zookeeper-ri
ports:
- "2181:2181"
kafka1:
image: confluenti
浏览 10提问于2022-06-14得票数 0
1回答
我有一个要求，从多个卡夫卡topicsAvro的流媒体，并把他们在格林梅利与小修改的有效载荷。
Kaka主题定义为配置文件中的列表，每个Kafka主题都有一个目标表。
我正在寻找一个单一的星火结构应用程序和一个更新的配置文件，以听取新的主题或停止。听这个话题。
我在寻求帮助，因为我对使用单个查询还是多个查询感到困惑：
val query1 = df.writeStream.start()
val query2 = df.writeStream.start()
spark.streams.awaitAnyTermination()
或
df.writeStream.start().awaitA
浏览 2提问于2020-11-06得票数 1
回答已采纳
1回答
使用聚合平台，Apache Kafka的创建者提供的平台，我有一个问题：
在的文档中，他们提到了“主题”的抽象。您可以在"subject“下注册一个模式，该模式的形式为topicName-key，或topicName-value，，但对于为什么需要(顾名思义)特定主题上的消息的键和值没有任何解释。也没有任何直接声明的大意是，注册与“主题”必然关联的模式与该主题，除了助记符。
更令人困惑的是，该页面上的后续示例(“获取主题模式版本”和“在subject下注册新模式”)不使用该格式作为主题名称，而只是使用主题名称作为"subject“值。如果有人有任何洞察力，a)为什么会有这两个“
浏览 1提问于2015-05-11得票数 12
我们需要从Kafka中提取数据并写入AWS s3。Kafka由单独的部门管理，我们只能访问特定的主题。
根据Kafka文档，看起来Kafka Connect对我来说是一个简单的解决方案，因为我没有任何自定义的消息处理逻辑。
通常，当我们运行Kafka Consumer时，我们可以使用相同的使用者组运行多个JVM，以实现可伸缩性。特定消费者的消费者JVM可以运行在同一台物理服务器上，也可以运行在不同的物理服务器上。当我想使用Kafka Connect时，会出现什么情况？
假设我有20个主题分区。
如何运行20个实例的Kafka Connect？
是否可以在同一物理实例上运行多个Kafka Con
浏览 2提问于2018-03-08得票数 1
我正在尝试创建一个具有多个处理器的Kafka Stream Topology，这些处理器订阅了正则表达式模式和多个主题。
例如，pattern topic_(?:a|b|c)将连接到topic_a、topic_b和topic_c。我对每个主题都有不同数量的分区，因为它们有不同的负载。但是，在运行我的拓扑时，我看到：
org.apache.kafka.streams.errors.TopologyException
...
Topics not co-partitioned
它们之间的区别是：
topic_a: 3 partitions
topic_b: 3 partitions
topic_c
浏览 10提问于2021-09-24得票数 0
我们有一个带有多个avroregistry的KafkaAvroSerde。在某个时候，serde尝试在1 url上注册模式时超时，但是由于它向流应用程序抛出IO异常，流线程关闭了。从kafka流应用程序的角度来看，这种类型违背了在创建avro serdes时支持多个urls的功能，因为DSL堆栈中出现的运行时异常将关闭stream Thread。几个问题：
有什么好办法处理这件事吗？
我们是否需要在应用程序逻辑中执行重试(当您简单地将主题具体化到商店中时，这可能是很棘手的)？
否则是否有一个avroserde包装器
可以用实际配置的avroRegistry urls重试吗？
当
浏览 0提问于2018-11-02得票数 2
回答已采纳
我有一个Kafka Consumer，目前配置了：
kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
但我真正想要的是能够使用Kryo反序列化程序：
public class KryoPOJODeserializer<T> implement
浏览 1提问于2016-08-15得票数 2
回答已采纳
我有一个接收器连接器，配置如下
{
"name": "sink-test-mariadb-MY_TOPIC",
"config": {
"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max":"10",
"topics":"MY_TOPIC",
浏览 0提问于2019-01-17得票数 4
我已经创建了一个NiFi流，它最终将json记录发布为带有Avro编码值和字符串键的记录，使用模式中的模式。这是 in NiFi。
我现在试图使用Kafka (connect-standalone)使用JdbcSinkConnector将消息移动到PostgreSQL数据库，但得到了以下错误：
我已经确认我的为1的架构。
Worker Config:
bootstrap.servers=localhost:29092
key.converter.schemas.enable=false
key.converter=org.apache.kafka.connect.storage.StringC
浏览 0提问于2018-12-11得票数 0
回答已采纳
我正在试用Spark结构流与Kafka。我正在寻找这个强制性的选项subscribePatternJava正则表达式字符串的卡夫卡选项。显然只有3个值是可能的：“指定”、“订阅”或"subscribePattern“
当我在googled上搜索这个选项时，最有用的信息是：
有人能给我加上一个外行术语吗?在这三种选择中，最明显的区别是什么？对于Spark来说，不同的行为是什么？
浏览 0提问于2017-02-22得票数 1
回答已采纳
我是KAFKA协议世界的新手，我想问你一些与我的项目相关的重要信息。
我正在使用AVRO文件来生产和消费消息，我想知道是否可以将相同的avro文件用于多个主题，例如，通过在生产者中使用不同的"name“属性，以及在使用者中使用特定的"name”属性。
非常感谢。斯特凡诺
浏览 0提问于2018-07-20得票数 0
我将迁移到kafka作为代理和debezium，以便从报告和搜索数据库的所有微服务中获取数据(ETL数据)。
有没有办法配置debezium，使其根据自定义条件(如用户或公司或行/数据的某些关键列/属性)将数据放在不同的主题上。
浏览 0提问于2019-01-24得票数 1
2回答
我有一个spring处理器模块，带有errorChannel和outputChannel的http出站网关.使用HTTP200的任何消息都会发送到outputChannel，其余消息都在failureChannel中。
现在，http处理器模块用TopicX的kafka出站适配器连接到Kafka-Sink .TopicX只接收HTTP200个消息以供进一步处理。现在，我们需要将failureChannel中的消息路由到TopicY。
如何在卡夫卡水槽中发送多个卡夫卡主题信息。消息头中有httpStatusCode。我的项目中使用的Kafka版本为0.8.2，java版本为1.7
<!--
浏览 5提问于2017-03-28得票数 2
回答已采纳
关于Kafka和schemas如何处理Avro序列化消息，我在中了解了模式ID存储在每个消息中的一个可预测的位置，因此我们似乎可以在相同的主题中获得不同模式的消息，并能够找到正确的模式并在此基础上成功地反序列化它们。另一方面，我看到许多人似乎在使用表达式“一个附加在主题上的模式”，但这意味着每个主题都有一个模式。
那哪个是对的？我能否利用模式注册中心(如KSql)的优势，并在同一主题中包含不同类型(或多个模式)的消息？
浏览 0提问于2019-01-21得票数 5
回答已采纳
1回答
我正在考虑将Kafka作为最终将向用户发送推送通知的worker节点的消息代理的几种技术之一。换句话说，我希望每个租户以大致相同的速率处理他们的消息。我的选择似乎要么为每个租户创建一个主题，要么为每个租户创建一个分区。两者似乎都是有问题的和/或不受欢迎的。
为每个租户创建一个主题似乎是一场后勤噩梦。每次将新租户添加到应用程序中时，都必须以某种方式通知使用者订阅该主题。
为每个租户创建一个分区看起来并不是很糟糕，但似乎不太受欢迎。然而，根据我对如何在分区和使用者之间分配负载的理解，如果多个租户共享同一个分区，则一个租户的消息可能会被卡在另一个租户的消息后面，这不是我想要的负载平衡方式。
我最好的
浏览 0提问于2019-05-02得票数 2
我正在创建一个CDC Postgres.source --> Kafka / Debezium --> Postgres.sink。 源主题创建得很好，但只有在表有数据的情况下才能创建。有没有办法在没有数据的情况下实例化所有的主题？因为我想用maven命令下载所有AVRO模式： mvn schema-registry:download 我的配置： {
"name": "postgres-source",
"config": {
"connector.class":"io.debezium.co
浏览 20提问于2021-07-30得票数 0
我想将流SQL放在Kafka中，以供Flink for CEP使用。这是一种好方法吗？
我知道动态模式定义在Flink上是不允许的，我需要应用它们可以在无界事件流上更改的规则。
举个例子；
有一个供用户定义其设备规则的UI。想象一下，有一个事物跟踪服务(如亚马逊网络服务中心)保持物理设备的状态，我也认为这是将每个设备的特定规则放入影子规则中的好方法，当影子参与者接收到传感器数据时，它可以发出数据，并将其规则添加到IoT作为规则引擎使用(通过kafka)。因此，我希望Flink执行传入的传感器数据(及其规则)，这些数据对于每个设备都可能是不同的。
浏览 7提问于2017-07-20得票数 1
我有一个运行Zookeeper、Confluent Schema注册表和Kafka安全管理器(KSM)的Kafka集群。KSM，，is软件使得使用csv文件而不是使用命令行工具来管理Kafka ACL变得很容易。
confluent模式注册表允许我们存储Kafka的Avro模式。它目前是开放的，我需要确保它的安全。我想给每个用户只读或获取权限。我目前使用kubernetes来部署所有的工具。
我如何使用KSM做到这一点？我在哪里可以找到示例？
谢谢
浏览 19提问于2021-04-20得票数 0
1回答
要明确的是，我并不是试图使用Kafka作为事件来源的数据存储库，仅仅是为了复制事件。
卡夫卡的看起来非常有趣，因为它可以验证生产者发送到某个主题的消息的模式。但是，据我所知，它将每个主题视为容器文件--每个主题都有一个模式。
此限制不适用于事件源流，对于单个聚合(如File )，您将有多个消息模式：FileCreated、FileMoved、FileCopied、FileDeleted。将其中的每一个都放在一个单独的主题上会很复杂，而且容易出错。
是否存在像schemas这样的工具，它支持同一主题的多个模式？
更新
为了澄清，上面的每个消息都有一个不同的模式。例如：
FileCreated
{
浏览 4提问于2016-06-28得票数 2
回答已采纳
1回答
我一直试图将数据从SQL server (启用更改跟踪)加载到Kafka，以便一个或多个系统(报表、其他DB等)能够使用这些数据。
我已经成功地为sql server(confluentinc/kafka-connect-cdc-mssql:1.0.0-preview)配置了卡夫卡连接插件，我还成功地在卡夫卡机器上启动了它。
我一直在寻找帮助回答以下问题的文档(找不到)
我如何将卡夫卡的主题与这种联系联系起来？
根据我发现的信息(在debezium论坛上)，将为每个表创建一个主题-->它是否与kafka sql server连接器的工作方式相同？
我在分布式模式下配置了连接，
浏览 0提问于2018-11-08得票数 2
1回答
我正在尝试使用Replicator as it is specified in the tutorial将模式注册表从本地集群迁移到Confluent Cloud。它成功地复制了所有的主题，除了几个，我不知道为什么…它会显示以下错误： ERROR Failed to translate schema registry record (io.confluent.connect.replicator.schemas.SchemaTranslator:188)
io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientEx
浏览 14提问于2020-10-06得票数 0
我正在评估Azure Event Hub与Kafka作为Service Broker的使用。我希望我能够同时创建两个本地应用程序，一个使用Kafka消费消息，另一个使用Azure Event Hub。我已经设置了一个docker容器，它是一个Kafka实例，我正在使用我的Azure帐户设置Azure事件中心(据我所知，没有其他方法可以为Azure事件中心创建本地/开发实例)。
有没有人有任何关于这两者的信息，在比较它们的功能时可能会有用？
浏览 35提问于2020-09-18得票数 6
1回答
我有一个场景，在这个场景中，我需要在Kafka中导入一个完整的DB，并在DB术语中创建一些用户可以查询的表的视图。我的要求是通过物理模型(表)中的视图重新构建逻辑模型。
因此，我想知道这样做的步骤。
我的理想是，kafka创建与表相对应的主题，然后由我声明性地(使用KSQL)创建视图。
虽然我在这里描述的东西一开始听起来是可行的，但是我对数据--主题中的数据的结构(模式)--有一个问题。问题似乎是，我可能不得不做一个额外的步骤，但不知道它是否可以避免或实际上是必要的。
更具体地说，视图通常表示表上的联接。我想，如果我想要在表上加入，我需要已经创建了Ktable或Kstream，这给出了要在其上
浏览 2提问于2019-07-06得票数 0
回答已采纳
当Kafka中的主题不存在，并且使用TopicEndpoint方法对其进行端点配置时，它会抛出一个异常：Confluent.Kafka.ConsumeException: Broker: Unknown topic or partition。在创建主题之后，异常不再发生。如何设置使用TopicEndpoint方法配置的端点的自动主题创建？
浏览 2提问于2021-01-07得票数 0
回答已采纳
1回答
我将数据和值模式发送给卡夫卡主题，如：：
./bin/kafka-avro-console-producer \
--broker-list 10.0.0.0:9092 --topic orders \
--property parse.key="true" \
--property key.schema='{"type":"record","name":"key_schema","fields":[{"name":"id","type&#
浏览 2提问于2020-04-14得票数 0
回答已采纳
我使用MongoSourceConnector将kafka主题与mongo数据库集合连接起来。对于带有单个kafka主题的单个数据库，它工作得很好，但是有什么方法可以用单个kafka主题连接多个mongo数据库吗？
浏览 3提问于2021-11-25得票数 0
我有一个要求建立聚合的数据，我们收到我们的Apache Kafka… 我有点不知道该走哪条技术路线…… 似乎人们看到的是标准的方式，一群Apache Kafka <-> Apache Spark <-> Solr Bitnami Data Platform 我找不到具体的例子说明这是如何工作的，但我也在问自己，是否有任何解决方案 Apache Kafka <-> Kafka Connect Solr <-> Solr 不会这样做，因为solr也支持聚合... Solr Aggregation 但我看到了一些代码片段，它们将数据聚合在Spark中
浏览 34提问于2021-11-11得票数 0
1回答
目前，我正试图围绕卡夫卡消费者解决方案建立一些集成测试。对于同质主题，我已经做得很好了，但是测试也要求我设置一个异构的主题，而且我似乎不能正确地指定模式。
在控制面板中，我所做的是创建一个记录模式数组，如下所示：
[
{
"fields": [
{
"name": "foo",
"type": "string"
}
],
"name": "TestRecord",
"namespace
浏览 4提问于2022-09-26得票数 0
1回答
我有一个卡夫卡主题，我通过卡夫卡制作人发送数据。现在在消费者方面，我有两种选择。
1.使用KafkaConsumer - kafkaConsumer代码如下所示，它从主题中读取数据及其工作状态.
@EnableKafka
@Configuration
@PropertySource("kaafka.properties")
public class RawEventKafkaConsumer {
private static final Logger logger = LoggerFactory.getLogger(RawEventKafkaConsumer.cl
浏览 0提问于2019-04-23得票数 3
回答已采纳
我有一个kafka流应用程序，它在其中使用stateStore (由RocksDB支持)。
流线程所做的一切就是从kafka主题中获取数据并将数据放入状态存储。(还有其他线程从statestore读取数据并进行业务逻辑处理)。
我观察到，由于stateStore，它创造了一个新的卡夫卡主题"changelog“。
但是我没有明白“改变”卡夫卡主题的目的是什么？
为什么需要它？，statestore和"changelog“kafka主题之间的关系是什么？是谁把数据放入这个主题？("changelog")
浏览 0提问于2020-05-01得票数 5
使用汇合5.4.1
我们碰巧在将连接流从KTable KTable转发到另一个主题时，在新的KTable外键连接中遇到了一个问题。
模式注册表和Avro序列化器相关构件:合流版本5.4.1 kafkastreams客户端: 5.4.1-ccs
外键跨ktable连接之后，虽然使用“for each”从ktable消费流，但如果使用“to”将联接流转发到另一个主题，则会引发模式注册表错误。
将错误中提到的模式与在模式注册表中注册的模式进行比较，它是完全相同的…。似乎卡夫卡2.4.0已经有了类似的问题：和这个问题继续存在，同时将它转发到另一个主题
Caused by: io.confluent.k
浏览 5提问于2020-03-16得票数 0
回答已采纳
1回答
我有一个生产者，它将一个json文件写入到主题中，以供kafka消费者流读取。它的简单键值对。我想流式传输主题，并通过添加/连接更多JSON键值行来丰富事件，然后发布到另一个主题。顺便说一下，这些值或键都没有任何共同之处。我可能想得太多了，但是我该如何绕过这个逻辑呢？
浏览 0提问于2020-07-31得票数 1
我们正在使用Kafka在数据库之间同步表(Debezium对此是完美的，但这是不可能的)。
Sync在一般情况下运行良好，但是在主题中存储的事件/消息的数量似乎是预期的3倍。
这是什么原因？
一些附加信息
目标数据库包含消息的确切数量(主题/3中的消息计数)。
大部分主题被分成3个分区(键通过SMT设置，使用DefaultPartitioner )。
JDBC源连接器
{
"name": "oracle_source",
"config": {
"connector.class": "io.confluen
浏览 1提问于2019-12-16得票数 0
1回答
使用Bash shell，是否有一种方法可以删除并重新创建运行命令时显示的所有Kafka主题？
kafka-topics --zookeeper localhost:2181
浏览 3提问于2022-01-05得票数 -1
1回答
我们有一个用例，我们试图使用多个Kafka主题(、AVRO、)与模式注册表集成。我们使用星火结构流(火花版本: 2.4.4)，合流卡夫卡(图书馆版本: 5.4.1)进行相同的操作：
val kafkaDF: DataFrame = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "<bootstrap.server>:9092")
.option("subscribe", [List of Topics]) // mu
浏览 5提问于2020-04-18得票数 0
给定一个配置为订阅多个主题的DataStreamReader (请参阅)：
// Subscribe to multiple topics
spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "topic1,topic2,topic3")
当我在此基础上使用foreachBatch时，批将包含哪些内容？
每个批次只
浏览 0提问于2019-07-11得票数 4
回答已采纳
我正在使用kafka来收集日志，是否还有其他分区的优点，比如我们以后是否可以使用它来简化日志处理，如果我们在一个主题中使用分区，
Partition增加了并行性，这是在kafka主题中使用partition的一个优点。
浏览 3提问于2015-09-24得票数 0
我有一个星火结构的流作业，它是基于subscribePattern的来自多个Kafka主题的流数据，对于每个Kafka主题，我都有一个火花模式。当从Kafka流数据时，我想根据主题名称将Spark模式应用于Kafka消息。
假设我有两个主题：cust & customers。
基于subscribePattern (Java )的Kafka流数据：
var df = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2
浏览 3提问于2020-04-29得票数 0
我有模式注册表容器http://registry-server:8081
ProducerConfig
bootstrap.servers : [PLAINTEXT://kafka-server:9092]
value.serializer : class org.apache.kafka.common.serialization.ByteArraySerializer
以及充当生产者的Standalone service，其属性设置如下
生产者
"schema.registry.url", "http://registry-server:80
浏览 7提问于2022-09-02得票数 1
回答已采纳
我试图向kafka发布/使用我的java对象。我使用Avro模式。
我的基本程序很好。在我的程序中，我在生产者(编码)和使用者(解码)中使用我的模式。
如果我在接收端将不同的对象发布到不同的主题(例如:100个主题)，我不知道我收到了什么类型的消息？我想从接收到的字节中获取avro模式，并将其用于解码。我的理解正确吗？如果是这样，如何从接收到的对象中检索？
浏览 1提问于2016-04-15得票数 0
回答已采纳
我已经检查了两个类似的问题，但它没有帮助。
node 1 [/appl/node1/confluent-4.0.0] ./bin/confluent status elasticsearch-sink
{"name":"elasticsearch-sink","connector":{"state":"RUNNING","worker_id":"10.192.226.24:8083"},"tasks":[{"state":"FAILED&#
浏览 2提问于2018-04-17得票数 0
我希望从配置中为TestContainers Kafka映像指定一个自定义端口，以便以后能够重新使用引导服务器param进行黑匣子应用程序测试，使用 Kafka模块没有找到用于在运行容器时修复端口的API，我发现的只是动态地将端口分配给容器。
class KafkaSetUpSpec extends AnyFlatSpec with TestContainerForAll with Matchers {
override val containerDef: KafkaContainer.Def = KafkaContainer.Def()
import org.testcontain
浏览 6提问于2022-05-16得票数 1
用例是使用kafka连接器同时写入多个表。
在第一个注释之后进行编辑
"insert.mode": "upsert",
"delete.enabled": "true",
"topics": "cars.best, airplanes.best",
"table.name.format": "CARS, AIRPLANES",
"pk.mode": "record_key",
"pk.fields": "cars
浏览 4提问于2022-07-20得票数 0
我有一个弹簧引导项目，我是spring连接到底层的kafka事件中心。
在同一个消费类中，我必须听两个不同的主题。我有两种方法可以这样做。
一个是有两个卡夫卡听众这样的：
@KafkaListener(topics = "topic1")
public void consumeTopic1(String message) throws Exception {
//do something
}
@KafkaListener(topics = "topic2")
public void consumeTopic2(String message)
浏览 1提问于2019-05-08得票数 0
我需要一个流来对主题A中的值进行分组，将分组后的值发送到主题B，并通过翻滚时间窗口将这些分组的值的总和发送到主题C。在Kafka中可以做到这一点吗？或者一个流只能从一个主题中读取和写入？
浏览 6提问于2018-08-08得票数 1
回答已采纳
我正在开发一个Kafka流应用程序，它将读取来自输入Kafka主题的消息，并过滤不必要的数据并将其推送到输出Kafka主题。
Kafka流配置：
@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
Map<String, Object> streamsConfiguration = new HashMap<>();
streamsConfig
浏览 0提问于2019-05-27得票数 8
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
