Spring Cloud Samples 再升级：Kafka 4.0 Share Groups + Stream 6 大场景 + RAG 独立模块，消息驱动与 AI 能力全面进化

Spring Boot 4.1|Kafka 4.0|Share Groups (KIP-932)|事务消息|RAG 独立模块|PgVector Redis在之前的博文中我们介绍了 spring-cloud-samples 项目的 AI 智能体、gRPC 通信和 Seata 分布式事务三大模块。时隔数月项目迎来了又一轮重要升级cloud-kafka-sample—— 基于 Kafka 4.0 的全新消息模块演示 Share Groups (KIP-932)、事务消息、3 节点集群高可用cloud-stream-sample—— Spring Cloud Stream RocketMQ 6 大场景深度演示从基础消费到延迟/顺序/事务消息cloud-ai-rag-sample—— RAG 检索增强生成独立模块化支持 PgVector Redis 双向量库切换下面逐一详解。 模块一cloud-kafka-sample —— Kafka 4.0 Share Groups 与事务消息实战为什么要加 Kafka 模块Kafka 作为分布式消息系统的标杆在微服务架构中扮演着至关重要的角色。然而Kafka 4.0 引入了Share Groups (KIP-932)突破了传统消费者组一个分区只能被一个消费者消费的限制事务消息是保证分布式系统数据一致性的关键能力但实际项目中很少看到完整的示例大多数 Kafka 示例只演示基础收发缺乏生产级的高级特性演示cloud-kafka-sample 正是为了填补这些空白。技术栈组件版本Spring Boot4.1.0Kafka4.3.1 (3 节点集群)端口8768功能全景1️⃣ 传统消费者组 —— 基础消息收发最经典的 Kafka 使用场景一个生产者一个消费者组消息被组内消费者独占消费。ComponentpublicclassConsumer{KafkaListener(topics${app.kafka.topic})voidprocessMessage(SampleMessagemessage){log.info(Received sample message [{}],message);}}启动时自动发送一条测试消息BeanApplicationRunnerrunner(Producerproducer){returnargs-producer.sendTraditional();}验证方式curl-XPOST http://localhost:8768/kafka/traditional2️⃣ Share Groups (KIP-932) —— 突破分区限制的并行消费这是 Kafka 4.0 最重要的新特性之一。传统消费者组中一个分区同一时刻只能被一个消费者消费而 Share Groups 允许多个消费者从同一个分区并行消费不同的消息非常适合任务分发和工作队列场景。项目演示了两种确认模式隐式确认模式—— 方法正常返回自动 ACCEPT抛出异常自动 REJECTKafkaListener(topics${app.kafka.share-topic},containerFactoryimplicitShareKafkaListenerContainerFactory,groupId${app.kafka.share-group})voidprocessImplicit(ConsumerRecordString,SampleMessagerecord){log.info([Share-Implicit] Received: {} from partition {} offset {},record.value(),record.partition(),record.offset());}显式确认模式—— 手动控制每条消息的确认支持三种状态KafkaListener(topics${app.kafka.share-topic-explicit},containerFactoryexplicitShareKafkaListenerContainerFactory,groupId${app.kafka.share-group},concurrency5)voidprocessExplicit(ConsumerRecordString,SampleMessagerecord,ShareAcknowledgmentack){try{if(record.value().getId()%50){ack.release();// 临时失败消息将被重新投递return;}ack.acknowledge();// 处理成功}catch(Exceptione){ack.reject();// 永久失败不再重试}}配置类的关键在于使用ShareKafkaListenerContainerFactory和DefaultShareConsumerFactoryBeanpublicShareKafkaListenerContainerFactoryString,SampleMessageexplicitShareKafkaListenerContainerFactory(ShareConsumerFactoryString,SampleMessagefactory){varcfnewShareKafkaListenerContainerFactory(factory);cf.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);returncf;}验证方式# 发送 10 条 Share Group 隐式确认消息curl-XPOSThttp://localhost:8768/kafka/share/implicit?count10# 发送 10 条显式确认消息id%50 的消息会被 release 重投递curl-XPOSThttp://localhost:8768/kafka/share/explicit?count10这意味着Share Groups 让 Kafka 从分区独占进化为分区共享大幅提升了消费并发度特别适合任务分发场景。3️⃣ 事务消息 —— 原子性发送与回滚事务消息保证多条消息的发送是原子性的要么全部成功对消费者可见要么全部不可见。生产者配置—— 显式创建事务和非事务两个 KafkaTemplateBeanQualifier(txKafkaTemplate)publicKafkaTemplateObject,SampleMessagetxKafkaTemplate(){MapString,ObjectpropsnewHashMap();props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,kafka-sample-tx);ProducerFactoryObject,SampleMessagepfnewDefaultKafkaProducerFactory(props);returnnewKafkaTemplate(pf);}发送事务消息—— 使用executeInTransaction保证原子性publicvoidsendTransactional(intcount,booleancommit){try{this.txKafkaTemplate.executeInTransaction(operations-{for(inti1;icount;i){operations.send(this.txTopic,newSampleMessage(i,tx-task-i));}if(!commit){thrownewRuntimeException(Simulated rollback);}returnnull;});}catch(Exceptione){log.warn([TX] Transaction rolled back);}}消费者配置—— 使用read_committed隔离级别props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,read_committed);验证方式# 提交事务 —— 消费者收到消息curl-XPOSThttp://localhost:8768/kafka/tx/commit?count5# 回滚事务 —— 消费者收不到消息curl-XPOSThttp://localhost:8768/kafka/tx/rollback?count5这意味着事务消息让 Kafka 从至少一次进化为精确一次语义保证分布式系统的数据一致性。4️⃣ 3 节点 Kafka 集群 —— KRaft 高可用部署项目手动部署 3 节点 Kafka 集群KRaft 模式无 ZooKeeper下载 Kafka 安装包准备 3 份配置文件server-1/2/3.properties分别格式化存储后在独立终端启动生产者配置指向 3 个 Brokerspring:kafka:bootstrap-servers:localhost:9092,localhost:9094,localhost:9096这意味着3 节点集群演示了真实生产环境的高可用部署任何一个 Broker 宕机不影响消息收发。 模块二cloud-stream-sample —— Spring Cloud Stream 6 大场景为什么 Stream 模块值得单独介绍之前的博文只简单提到了 Stream 模块支持 RocketMQ 消息驱动但实际上它演示了6 个完整场景覆盖了消息中间件最常用的高级特性。技术栈组件版本Spring Boot4.1.0Spring Cloud Stream4.1.0RocketMQ5.x端口87676 大场景场景 1基础消费 —— Consumer 函数BeanpublicConsumerStringinput(){returnmessage-log.info(Received message: {},message);}场景 2定时消息源 —— Supplier 函数Supplier 每隔 1 秒自动生成一条消息BeanpublicSupplierStringoutput2(){return()-你好;}output2-out-0:destination:stream-demo-topic2producer:poller:fixed-delay:1000# 每秒发送一次场景 3消息处理管道 —— Function 函数Function 接收消息、转换后输出到新 TopicBeanpublicFunctionString,Stringtransform(){returnmessage-[PROCESSED] message.toUpperCase();}curl-XPOSThttp://localhost:8767/stream/send?messagehello# 日志消息转换: hello - [PROCESSED] HELLO场景 4延迟消息 —— RocketMQ 延迟级别通过 Message Header 设置延迟级别1-18消息在指定延迟后被消费MessageStringmsgMessageBuilder.withPayload(message).setHeader(DELAY,delayLevel).build();streamBridge.send(delayPublish-out-0,msg);curl-XPOSThttp://localhost:8767/stream/delay?messagehellodelayLevel2# 5 秒后消费者收到消息场景 5顺序消息 —— 分区键 顺序消费MessageStringmsgMessageBuilder.withPayload(message).setHeader(ORDER_KEY,orderKey).build();streamBridge.send(fifoPublish-out-0,msg);fifo-in-0:consumer:orderly:truefifoPublish-out-0:producer:orderly:truepartitionKeyExpression:headers[ORDER_KEY]curl-XPOSThttp://localhost:8767/stream/fifo?messagemsg1orderKeyorder-1curl-XPOSThttp://localhost:8767/stream/fifo?messagemsg2orderKeyorder-1# 消息按发送顺序被消费场景 6事务消息 —— RocketMQ 两阶段提交事务消息流程发送半消息 - 执行本地事务 - 根据结果 Commit/Rollback。Component(demoTransactionListener)publicclassDemoTransactionListenerimplementsTransactionListener{OverridepublicLocalTransactionStateexecuteLocalTransaction(Messagemsg,Objectarg){StringtxArgmsg.getProperty(TX_ARG);if(commit.equalsIgnoreCase(txArg))returnLocalTransactionState.COMMIT_MESSAGE;if(rollback.equalsIgnoreCase(txArg))returnLocalTransactionState.ROLLBACK_MESSAGE;returnThreadLocalRandom.current().nextBoolean()?LocalTransactionState.COMMIT_MESSAGE:LocalTransactionState.ROLLBACK_MESSAGE;}}# 提交事务 —— 消费者收到消息curl-XPOSThttp://localhost:8767/stream/tx?messagehelloargcommit# 回滚事务 —— 消费者收不到消息curl-XPOSThttp://localhost:8767/stream/tx?messagehelloargrollback这意味着Stream 模块覆盖了延迟、顺序、事务等生产环境最常用的 6 大场景每个场景都有完整的代码和验证方式。 模块三cloud-ai-rag-sample —— RAG 独立模块化为什么 RAG 要独立成模块在之前的博文中RAG 功能集成在 cloud-ai-sample 中。但随着 RAG 在企业级应用中的广泛落地它已经足够复杂和重要值得独立成一个模块向量数据库选型支持 PgVector 和 Redis 两种主流方案通过 Profile 切换知识库管理文档摄入、检索、删除的完整生命周期独立部署RAG 服务可以独立扩展技术栈组件版本Spring Boot4.1.0Spring AI2.0.0向量数据库PostgreSQL pgvector 或 RedisEmbedding 模型text-embedding-v3 (1024 维)端口8889核心功能1️⃣ 文档摄入 —— 自动分块与向量化publicintingest(Stringcontent,Stringsource){DocumentdocnewDocument(content);doc.getMetadata().put(source,source);ListDocumentchunkstextSplitter.split(doc);vectorStore.add(chunks);returnchunks.size();}curl-XPOST http://localhost:8889/ai/rag/ingest\-HContent-Type: application/json\-d{content:Spring AI 是 Spring 生态的 AI 集成框架...,source:docs}2️⃣ RAG 查询 —— 检索增强 LLM 回答publicStringquery(Stringquestion,inttopK){ListDocumentdocsvectorStore.similaritySearch(SearchRequest.builder().query(question).topK(topK).build());Stringcontextdocs.stream().map(Document::getText).collect(joining(\n));Stringprompt基于以下参考资料回答\ncontext\n\n问题question;returnchatClient.prompt().user(prompt).call().content();}curlhttp://localhost:8889/ai/rag/query?questionSpring AItopK33️⃣ 双向量库 —— PgVector 与 Redis 一键切换# 使用 PgVector默认java-jarcloud-ai-rag-sample.jar# 使用 Redisjava-jarcloud-ai-rag-sample.jar--spring.profiles.activeredis这意味着RAG 模块独立部署、双向量库支持企业可根据现有技术栈灵活选型。 模块总览16 个模块模块端口说明状态cloud-gateway-sample8764Gateway Sentinel 限流原有cloud-provider-sample8765Web 服务提供者原有cloud-consumer-sample8766Web 服务消费者原有cloud-provider-reactive-sample8762Reactive Web 提供者原有cloud-consumer-reactive-sample8763Reactive Web 消费者原有cloud-provider-dubbo-sample50051Dubbo RPC 提供者原有cloud-nacos-config-sample8761Nacos 动态配置原有cloud-stream-sample8767Stream6 大场景 增强cloud-ai-sample8888Spring AI 2.0 全场景原有cloud-grpc-server-sample9090gRPC Server原有cloud-seata-sample18081-84Seata 分布式事务原有cloud-kafka-sample8768Kafka 4.0 新增cloud-ai-rag-sample8889RAGPgVector/Redis 独立cloud-nacos-discovery-sample8760Nacos 服务发现原有cloud-sample-api-接口定义 Proto原有cloud-commons-gRPC 服务发现桥接原有 学习价值cloud-kafka-sampleKafka 4.0 Share Groups (KIP-932) 隐式/显式确认事务消息原子性发送与 read_committed 隔离3 节点 Kafka 集群KRaft 模式高可用非事务与事务 KafkaTemplate 分离设计cloud-stream-sample函数式编程模型Consumer/Function/Supplier延迟消息RocketMQ 18 级延迟顺序消息分区键 orderly事务消息两阶段提交StreamBridge 编程式发布cloud-ai-rag-sampleRAG 完整流程摄入 - 分块 - 向量化 - 检索 - 增强PgVector 与 Redis 双向量库 Profile 切换基于元数据的文档过滤与删除 快速体验Kafka 模块# 1. 手动部署 3 节点 Kafka 集群KRaft 模式# 下载 Kafka 安装包参考 cloud-kafka-sample/README.md 准备 3 份配置并依次启动KAFKA_CLUSTER_ID$(bin/kafka-storage.sh random-uuid)bin/kafka-storage.shformat-t$KAFKA_CLUSTER_ID-cconfig/server-1.properties bin/kafka-storage.shformat-t$KAFKA_CLUSTER_ID-cconfig/server-2.properties bin/kafka-storage.shformat-t$KAFKA_CLUSTER_ID-cconfig/server-3.properties bin/kafka-server-start.sh config/server-1.properties# 终端1bin/kafka-server-start.sh config/server-2.properties# 终端2bin/kafka-server-start.sh config/server-3.properties# 终端3# 2. 启动 Kafka 模块cdcloud-kafka-samplemvn spring-boot:run# 3. 验证curl-XPOST http://localhost:8768/kafka/traditionalcurl-XPOSThttp://localhost:8768/kafka/share/implicit?count10curl-XPOSThttp://localhost:8768/kafka/tx/commit?count5curl-XPOSThttp://localhost:8768/kafka/tx/rollback?count5Stream 模块# 1. 启动 RocketMQbin/mqnamesrvbin/mqbroker-nlocalhost:9876# 2. 启动 Stream 模块cdcloud-stream-samplemvn spring-boot:run# 3. 验证 6 大场景curl-XPOSThttp://localhost:8767/stream/send?messagehellocurl-XPOSThttp://localhost:8767/stream/delay?messagehellodelayLevel2curl-XPOSThttp://localhost:8767/stream/fifo?messagemsg1orderKeyorder-1curl-XPOSThttp://localhost:8767/stream/tx?messagehelloargcommitRAG 模块# 1. 启动 PostgreSQL pgvectorpsql-Upostgres-cCREATE DATABASE ai_demo;psql-Upostgres-dai_demo-finit_ai_demo.sql# 2. 启动 RAG 模块cdcloud-ai-rag-samplemvn spring-boot:run# 3. 验证curl-XPOST http://localhost:8889/ai/rag/ingest\-HContent-Type: application/json\-d{content:Spring AI 支持 Chat 和 Embedding...,source:docs}curlhttp://localhost:8889/ai/rag/query?questionSpring AItopK3 最佳实践总结1. Kafka Share Groups vs 传统消费者组特性传统消费者组Share Groups分区消费一个分区只能被一个消费者消费多个消费者可并行消费同一分区并发度受限于分区数不受分区数限制适用场景流式处理、事件溯源任务分发、工作队列确认模式自动/手动 offset 提交隐式/显式 ACK2. Kafka 事务消息关键配置生产者必须配置transactional.id消费者必须设置isolation.levelread_committed使用executeInTransaction()保证原子性非事务和事务 KafkaTemplate 应分离3. Spring Cloud Stream 函数式模型Consumer终端消费者处理消息后不输出Function消息处理管道转换后输出到新 TopicSupplier定时消息源自动生成消息StreamBridge编程式消息发布4. RAG 的 temperature 调优RAG 场景建议temperature0.2忠于检索事实通用对话可用temperature0.7发挥创造力 相关链接项目地址: https://github.com/javahongxi/spring-cloud-samplesApache Kafka: https://kafka.apache.orgSpring Cloud Stream: https://spring.io/projects/spring-cloud-streamSpring AI: https://spring.io/projects/spring-ai️PgVector: https://github.com/pgvector/pgvector 结语从最初的 10 个模块到如今的 16 个spring-cloud-samples 始终围绕一个目标提供生产环境可参考的微服务全栈示例。本次新增的三大模块分别对应了消息驱动和 AI 工程化领域最核心的能力Kafka 4.0—— Share Groups 突破分区限制事务消息保证数据一致性Stream 6 大场景—— 从基础消费到延迟/顺序/事务覆盖生产环境最常用的消息模式RAG 独立模块—— 双向量库支持企业级知识库检索增强生成方案如果你正在 评估 Kafka 4.0 的新特性Share Groups、事务消息 学习 Spring Cloud Stream 的高级用法 构建企业级 RAG 应用 寻找一站式 Spring Cloud 学习项目Star ⭐ 这个项目三大模块即刻上手gitclone https://github.com/javahongxi/spring-cloud-samples.gitcdspring-cloud-samples© hongxi.org| 以生产环境可参考为目标持续打造最完整的 Spring Cloud 示例项目