不同@StreamListener的两个实例间的嵌入式Kafka迁移状态存储
是指在使用Spring Cloud Stream框架进行消息驱动的微服务开发中，当存在多个实例同时监听同一个Kafka主题时，如何保证消息的有序性和可靠性。
嵌入式Kafka是指将Kafka消息中间件直接集成到应用程序中，而不是使用外部的Kafka集群。在这种情况下，每个实例都会创建一个嵌入式Kafka实例，用于接收和处理消息。
迁移状态存储是指在实例之间迁移状态信息，以便实现消息的有序处理和故障恢复。当一个实例处理完一条消息后，它会将处理的状态信息存储到一个共享的存储介质中，以便其他实例可以获取到这个状态信息，并在此基础上进行处理。
为了实现不同@StreamListener实例间的嵌入式Kafka迁移状态存储，可以采用以下步骤：
- 使用Spring Cloud Stream框架创建多个实例，每个实例都使用相同的Kafka主题进行监听。
- 在每个实例中，使用嵌入式Kafka实例来接收和处理消息。
- 在每个实例中，使用一个共享的存储介质（如数据库、Redis等）来存储处理的状态信息。
- 当一个实例处理完一条消息后，将处理的状态信息存储到共享的存储介质中。
- 其他实例可以通过读取共享的存储介质来获取到最新的状态信息，并在此基础上进行处理。
这种方式可以保证不同实例之间的消息处理的有序性和可靠性。当一个实例发生故障时，其他实例可以通过读取共享的存储介质来获取到最新的状态信息，并继续处理未完成的消息。
在腾讯云的云计算平台中，可以使用腾讯云的消息队列CMQ作为共享的存储介质，用于存储处理的状态信息。CMQ是一种高可用、高可靠的消息队列服务，可以满足消息处理的需求。相关产品介绍和链接地址如下：
产品名称：腾讯云消息队列 CMQ 产品介绍链接：https://cloud.tencent.com/product/cmq
相关·内容
我有一个带有通过Spring Cloud映射的两个流处理器的SpringBoot应用程序。每个处理器都有自己的@StreamListener用于不同的主题。一个处理器将聚集的数据写入可四分状态存储。当我通过@Service (服务从状态存储获取聚合数据)获取数据时，我在单元测试中遇到了这个问题。:60) at org.apache.kafka.stream
浏览 10提问于2018-09-08得票数 0
我遇到了一些困难，试图通过卡夫卡流实现以下目标：
val builder = new StreamsBuilderS()
Stores.persistentKeyValueSto
浏览 3提问于2019-04-12得票数 3
我正在使用带有MockSchemaRegistryClient的为我们的kafka系统构建一个集成测试。我正在为我们的一个Stream拓扑构建一个测试，该拓扑是使用Streams API (KStreamBuilder)构建的。我在将来自表1的KTableProcessor的输入馈送到stream1时遇到错误：
Exception in thread "mortgage-kafka-consumers-it-c1dd9185-at org.a
浏览 2提问于2018-06-19得票数 5
我想了解一些关于如何将状态存储和主题分区分配给流处理应用程序及其任务的细节。
假设我有一个4分区主题(tA)。我还有4个实例(i0、i1、i2、i3)在4台不同的机器上运行相同的application.id (myApp)，以及来自tA的流记录。流引擎将为每个应用程序实例分配一个分区。为了便于讨论，我们假设分区分配是: p0->i0，p1->i1，p2->i2和p3->i3，并且假设我的流应用
浏览 5提问于2021-04-22得票数 1
回答已采纳
1回答
我正在Java应用程序()中使用Kafka。我有一个特定的用例，如下所示：由于应用程序的多个实例将运行，并且不能保证将两个主题的哪个特定分区分配给实例，因此必须在应用程序之间共享状态存储</
浏览 5提问于2020-02-20得票数 0
1回答
我有5个不同的机器与每个缩放5弹簧启动实例，使用kafka-streams应用程序。我使用50个不同的2-3个主题的分区压缩主题，每个实例有10个并发。当我重新部署我的实例，并用最新的实体填充来自elasticsearch的压缩主题时，我的连接操作是正常的。所以我认为当我的应用程序在新机器上启动时，我的本地<e
浏览 23提问于2020-01-07得票数 0
我们遇到了一个问题，因为接口从Kafka 0.11更改到Kafka 2.0。在我们基于0.11的Kafka streams应用程序中，我们有两个使用命名状态存储的KTables[String,Something]之间的连接： myClass1Table.leftJoinMyClass1,MyClass2](myClass2Table,new
MyJoiner, new M
浏览 1提问于2018-11-29得票数 0
1回答
我试图从单个输入流创建多输出流(取决于不同的时间窗口)。PAGE_VIEWS_COUNTS_OUT_Last_30_Minutes) }
// aggregate by Duration.ofMinutes(5)
@
浏览 2提问于2020-07-09得票数 0
回答已采纳
我们有以下问题： .to(stringSerde, stringSerde, "transaction-summary50");
它创建了一个由Kafka我担心的是，如果我们
浏览 4提问于2017-01-30得票数 3
回答已采纳
我正在创建一个本地商店，以保存来自特定主题components的所有更新。我不知道我做错了什么。还有其他方法可以从Stream创建存储吗？ Thread.sleep(100); }} Exception in thread "main" org.apache.kafka.streams.errors.InvalidStateStoreException: The state stor
浏览 0提问于2019-05-09得票数 0
回答已采纳
在使用连接到Kafka的Spring应用程序中，我尝试设置两个单独的流侦听器方法：
Exception in thread "test-d44cb
浏览 4提问于2018-10-15得票数 1
回答已采纳
让我们假设我有一个有状态的Kafka应用程序，它使用来自主题的3个分区的数据。目前，我有两个以上应用程序的实例正在运行。所以现在我想要添加新的实例来完全利用并行化。
据我理解，一旦启动新实例，就会发生重新平衡:一个分区( part1或part2 )和相应的本地状态存储将从现有实例迁移到新添加的
浏览 4提问于2021-11-03得票数 0
回答已采纳
2回答
我已经阅读了有状态流处理，如果我理解正确的话，RocksDB被用作键值存储的默认实现的一个主要原因是，与内存中的集合不同，它可以处理大于可用内存的数据，因为它可以刷新到磁盘。这两种类型的商店都可以生存应用程序重新启动，因为数据备份为一个卡夫卡主题。
但还有其他区别吗？例如，我注意到我的持久状态存储为每个主题分区创建了一些.log文件，但它们都是空的。简而言之，我想知道用内存中<em
浏览 1提问于2018-03-08得票数 12
回答已采纳
我有一个应用程序，它需要在分布在地理区域的实例的状态之间进行有趣的编排，同时需要一个可伸缩的分布式数据库。目前，我认为具有日志压缩的Kafka将满足我对状态维护和实例间消息交换的需要，而Cassandra将满足我对持久化数据的大量分布式读写的需要。然而，有相当多的数据是这样重复的:通过Kafka</em
浏览 6提问于2021-12-08得票数 0
回答已采纳
1回答
我正在学习使用Apache的事件驱动架构和流系统。我了解了事件源和CQRS，并有一些关于实现的基本问题。我知道卡夫卡有一个持久化的状态存储可以维护所需的状态，但是我能像卡桑德拉表一样依赖它吗？有尺寸consideration?吗？如果整个应用程序以及所有
浏览 7提问于2022-09-12得票数 0
回答已采纳
据我所知，在这种类型的流处理器中，它使用State Store维护某种状态。A->B->C；处理器B为有状态的，具有本地状态存储和changelog 启用。我正在使用低级别API。我注意到，当应用程序启动时(在不同的物理机器中有2个实例，num.stream.thre
浏览 2提问于2020-05-05得票数 2
回答已采纳
我的问题是-我有Kafka流处理微服务，它可以侦听多个主题。聚合来自多个主题的状态，并将其存储在状态存储中。我们将聚合后的消息提前发送到下游系统。下游系统会返回不同Kafka主题的其他消息。需要创建对我的所有微服务实例都可见的全局状态存储。有没有可能使用全局存储在karka中实现相同的
浏览 4提问于2021-01-09得票数 0
我正在尝试访问在同一个java程序中创建的inMemoryStore。但是，将异常作为“线程中的异常”org.apache.kafka.streams.errors.InvalidStateStoreException:返回，状态存储storeName可能已迁移到另一个实例当我使用persistentKeyValueStore时，它工作得很好，能够创建存储并返回值。:中的异常状态存储示例可能已<em
浏览 0提问于2019-07-12得票数 0
我在本地机器上运行多个kafka流消费者实例(2个实例)，每个实例都有自己的自定义本地存储，并且每个实例都有不同的名称。根据文档，如果其中一个实例宕机，那么kafka必须将死实例的存储同步到活实例的存储(如果我错了，请纠正我)。
我已经为两个实例配置了相同<
浏览 1提问于2019-06-21得票数 0
1回答
如何管理分片的本地存储？
、、、、
让我们假设有一个单一的消费群体(从卡夫卡的角度)。使用者组由20个服务实例副本组成。在基于某些属性(UUID)的这20个实例中，所有工作都是平衡的。每个实例管理自己的存储/状态/读取，而存储/状态/读则只包含属于该碎片的数据。所以有20个独立的仓库，每个副本一个。但是，如果扩大或减少这些服务，会发生什么呢？
浏览 3提问于2022-03-20得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
