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 正是为了填补这些空白。

技术栈

功能全景

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/traditional

2️⃣ Share Groups (KIP-932) —— 突破分区限制的并行消费

这是 Kafka 4.0 最重要的新特性之一。传统消费者组中，一个分区同一时刻只能被一个消费者消费；而 Share Groups 允许多个消费者从同一个分区并行消费不同的消息，非常适合任务分发和工作队列场景。

项目演示了两种确认模式：

隐式确认模式—— 方法正常返回自动 ACCEPT，抛出异常自动 REJECT：

@KafkaListener(topics="${app.kafka.share-topic}",containerFactory="implicitShareKafkaListenerContainerFactory",groupId="${app.kafka.share-group}")voidprocessImplicit(ConsumerRecord<String,SampleMessage>record){log.info("[Share-Implicit] Received: {} from partition {} offset {}",record.value(),record.partition(),record.offset());}

显式确认模式—— 手动控制每条消息的确认，支持三种状态：

@KafkaListener(topics="${app.kafka.share-topic-explicit}",containerFactory="explicitShareKafkaListenerContainerFactory",groupId="${app.kafka.share-group}",concurrency="5")voidprocessExplicit(ConsumerRecord<String,SampleMessage>record,ShareAcknowledgmentack){try{if(record.value().getId()%5==0){ack.release();// 临时失败，消息将被重新投递return;}ack.acknowledge();// 处理成功}catch(Exceptione){ack.reject();// 永久失败，不再重试}}

配置类的关键在于使用ShareKafkaListenerContainerFactory和DefaultShareConsumerFactory：

@BeanpublicShareKafkaListenerContainerFactory<String,SampleMessage>explicitShareKafkaListenerContainerFactory(ShareConsumerFactory<String,SampleMessage>factory){varcf=newShareKafkaListenerContainerFactory<>(factory);cf.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);returncf;}

验证方式：

# 发送 10 条 Share Group 隐式确认消息curl-XPOST"http://localhost:8768/kafka/share/implicit?count=10"# 发送 10 条显式确认消息（id%5==0 的消息会被 release 重投递）curl-XPOST"http://localhost:8768/kafka/share/explicit?count=10"

这意味着：Share Groups 让 Kafka 从"分区独占"进化为"分区共享"，大幅提升了消费并发度，特别适合任务分发场景。

3️⃣ 事务消息 —— 原子性发送与回滚

事务消息保证多条消息的发送是原子性的：要么全部成功对消费者可见，要么全部不可见。

生产者配置—— 显式创建事务和非事务两个 KafkaTemplate：

@Bean@Qualifier("txKafkaTemplate")publicKafkaTemplate<Object,SampleMessage>txKafkaTemplate(){Map<String,Object>props=newHashMap<>();props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"kafka-sample-tx");ProducerFactory<Object,SampleMessage>pf=newDefaultKafkaProducerFactory<>(props);returnnewKafkaTemplate<>(pf);}

发送事务消息—— 使用executeInTransaction保证原子性：

publicvoidsendTransactional(intcount,booleancommit){try{this.txKafkaTemplate.executeInTransaction(operations->{for(inti=1;i<=count;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-XPOST"http://localhost:8768/kafka/tx/commit?count=5"# 回滚事务 —— 消费者收不到消息curl-XPOST"http://localhost:8768/kafka/tx/rollback?count=5"

这意味着：事务消息让 Kafka 从"至少一次"进化为"精确一次"语义，保证分布式系统的数据一致性。

4️⃣ 3 节点 Kafka 集群 —— KRaft 高可用部署

项目手动部署 3 节点 Kafka 集群（KRaft 模式，无 ZooKeeper）：下载 Kafka 安装包，准备 3 份配置文件（server-1/2/3.properties），分别格式化存储后在独立终端启动，生产者配置指向 3 个 Broker：

spring:kafka:bootstrap-servers:localhost:9092,localhost:9094,localhost:9096

这意味着：3 节点集群演示了真实生产环境的高可用部署，任何一个 Broker 宕机不影响消息收发。

🔄 模块二：cloud-stream-sample —— Spring Cloud Stream 6 大场景

为什么 Stream 模块值得单独介绍？

之前的博文只简单提到了 Stream 模块支持 RocketMQ 消息驱动，但实际上它演示了6 个完整场景，覆盖了消息中间件最常用的高级特性。

技术栈

6 大场景

场景 1：基础消费 —— Consumer 函数

@BeanpublicConsumer<String>input(){returnmessage->log.info("Received message: {}",message);}

场景 2：定时消息源 —— Supplier 函数

Supplier 每隔 1 秒自动生成一条消息：

@BeanpublicSupplier<String>output2(){return()->"你好";}

output2-out-0:destination:stream-demo-topic2producer:poller:fixed-delay:1000# 每秒发送一次

场景 3：消息处理管道 —— Function 函数

Function 接收消息、转换后输出到新 Topic：

@BeanpublicFunction<String,String>transform(){returnmessage->"[PROCESSED] "+message.toUpperCase();}

curl-XPOST"http://localhost:8767/stream/send?message=hello"# 日志：消息转换: hello -> [PROCESSED] HELLO

场景 4：延迟消息 —— RocketMQ 延迟级别

通过 Message Header 设置延迟级别（1-18），消息在指定延迟后被消费：

Message<String>msg=MessageBuilder.withPayload(message).setHeader("DELAY",delayLevel).build();streamBridge.send("delayPublish-out-0",msg);

curl-XPOST"http://localhost:8767/stream/delay?message=hello&delayLevel=2"# 5 秒后消费者收到消息

场景 5：顺序消息 —— 分区键 + 顺序消费

Message<String>msg=MessageBuilder.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-XPOST"http://localhost:8767/stream/fifo?message=msg1&orderKey=order-1"curl-XPOST"http://localhost:8767/stream/fifo?message=msg2&orderKey=order-1"# 消息按发送顺序被消费

场景 6：事务消息 —— RocketMQ 两阶段提交

事务消息流程：发送半消息 -> 执行本地事务 -> 根据结果 Commit/Rollback。

@Component("demoTransactionListener")publicclassDemoTransactionListenerimplementsTransactionListener{@OverridepublicLocalTransactionStateexecuteLocalTransaction(Messagemsg,Objectarg){StringtxArg=msg.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-XPOST"http://localhost:8767/stream/tx?message=hello&arg=commit"# 回滚事务 —— 消费者收不到消息curl-XPOST"http://localhost:8767/stream/tx?message=hello&arg=rollback"

这意味着：Stream 模块覆盖了延迟、顺序、事务等生产环境最常用的 6 大场景，每个场景都有完整的代码和验证方式。

🧠 模块三：cloud-ai-rag-sample —— RAG 独立模块化

为什么 RAG 要独立成模块？

在之前的博文中，RAG 功能集成在 cloud-ai-sample 中。但随着 RAG 在企业级应用中的广泛落地，它已经足够复杂和重要，值得独立成一个模块：

• 向量数据库选型：支持 PgVector 和 Redis 两种主流方案，通过 Profile 切换
• 知识库管理：文档摄入、检索、删除的完整生命周期
• 独立部署：RAG 服务可以独立扩展

技术栈

核心功能

1️⃣ 文档摄入 —— 自动分块与向量化

publicintingest(Stringcontent,Stringsource){Documentdoc=newDocument(content);doc.getMetadata().put("source",source);List<Document>chunks=textSplitter.split(doc);vectorStore.add(chunks);returnchunks.size();}

curl-XPOST http://localhost:8889/ai/rag/ingest\-H"Content-Type: application/json"\-d'{"content":"Spring AI 是 Spring 生态的 AI 集成框架...","source":"docs"}'

2️⃣ RAG 查询 —— 检索增强 LLM 回答

publicStringquery(Stringquestion,inttopK){List<Document>docs=vectorStore.similaritySearch(SearchRequest.builder().query(question).topK(topK).build());Stringcontext=docs.stream().map(Document::getText).collect(joining("\n"));Stringprompt="基于以下参考资料回答：\n"+context+"\n\n问题："+question;returnchatClient.prompt().user(prompt).call().content();}

curl"http://localhost:8889/ai/rag/query?question=Spring AI&topK=3"

3️⃣ 双向量库 —— PgVector 与 Redis 一键切换

# 使用 PgVector（默认）java-jarcloud-ai-rag-sample.jar# 使用 Redisjava-jarcloud-ai-rag-sample.jar--spring.profiles.active=redis

这意味着：RAG 模块独立部署、双向量库支持，企业可根据现有技术栈灵活选型。

📊 模块总览：16 个模块

🎓 学习价值

cloud-kafka-sample

• Kafka 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-sample

• RAG 完整流程（摄入 -> 分块 -> 向量化 -> 检索 -> 增强）
• 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-sample&&mvn spring-boot:run# 3. 验证curl-XPOST http://localhost:8768/kafka/traditionalcurl-XPOST"http://localhost:8768/kafka/share/implicit?count=10"curl-XPOST"http://localhost:8768/kafka/tx/commit?count=5"curl-XPOST"http://localhost:8768/kafka/tx/rollback?count=5"

Stream 模块

# 1. 启动 RocketMQbin/mqnamesrv&&bin/mqbroker-nlocalhost:9876# 2. 启动 Stream 模块cdcloud-stream-sample&&mvn spring-boot:run# 3. 验证 6 大场景curl-XPOST"http://localhost:8767/stream/send?message=hello"curl-XPOST"http://localhost:8767/stream/delay?message=hello&delayLevel=2"curl-XPOST"http://localhost:8767/stream/fifo?message=msg1&orderKey=order-1"curl-XPOST"http://localhost:8767/stream/tx?message=hello&arg=commit"

RAG 模块

# 1. 启动 PostgreSQL + pgvectorpsql-Upostgres-c"CREATE DATABASE ai_demo;"psql-Upostgres-dai_demo-finit_ai_demo.sql# 2. 启动 RAG 模块cdcloud-ai-rag-sample&&mvn spring-boot:run# 3. 验证curl-XPOST http://localhost:8889/ai/rag/ingest\-H"Content-Type: application/json"\-d'{"content":"Spring AI 支持 Chat 和 Embedding...","source":"docs"}'curl"http://localhost:8889/ai/rag/query?question=Spring AI&topK=3"

💡 最佳实践总结

1. Kafka Share Groups vs 传统消费者组

2. Kafka 事务消息关键配置

• 生产者必须配置transactional.id
• 消费者必须设置isolation.level=read_committed
• 使用executeInTransaction()保证原子性
• 非事务和事务 KafkaTemplate 应分离

3. Spring Cloud Stream 函数式模型

• Consumer：终端消费者，处理消息后不输出
• Function：消息处理管道，转换后输出到新 Topic
• Supplier：定时消息源，自动生成消息
• StreamBridge：编程式消息发布

4. RAG 的 temperature 调优

• RAG 场景建议temperature=0.2，忠于检索事实
• 通用对话可用temperature=0.7，发挥创造力

🔗 相关链接

• 📦项目地址: https://github.com/javahongxi/spring-cloud-samples
• 📨Apache Kafka: https://kafka.apache.org
• 🔄Spring Cloud Stream: https://spring.io/projects/spring-cloud-stream
• 🧠Spring 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 示例项目