使用Reactive Cassandra无法存储Kafka接收的消息求助

问题背景 #

通过Kafka接收不同微服务的消息，使用Reactive Cassandra进行存储时功能异常，但将仓库从ReactiveCassandraRepository改为普通CassandraRepository后功能正常。

项目依赖 #

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-cassandra-reactive</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bus-kafka</artifactId>
</dependency>

当前代码 #

Repository定义 #

@Repository
public interface AuditLogRepository extends ReactiveCassandraRepository<AuditLog, String> {

}

消息处理与保存方法 #

@Autowired
private AuditLogRepository auditLogRepository;

public void saveAuditLog(Flux<AuditLogMessage> auditLogMessageFlux) {

    auditLogMessageFlux.subscribe(auditLogMessage -> {
        AuditLog audit = AuditLog.builder().id(UUID.randomUUID().toString()).tenantId(auditLogMessage.getTenantId())
                .logTime(auditLogMessage.getLogTime()).entity(auditLogMessage.getEntity())
                .userId(auditLogMessage.getUserId()).logType(auditLogMessage.getLogType())
                .entityReference(auditLogMessage.getEntityReference()).change(auditLogMessage.getChange()).build();
        auditLogRepository.save(audit);
    });
}

问题原因 #

核心问题在于Reactive编程模型的特性：

• ReactiveCassandraRepository的save方法返回Mono<AuditLog>，这是惰性操作，只有被订阅时才会实际执行数据库写入。
• 原代码仅调用auditLogRepository.save(audit)但未订阅返回的Mono，因此写入操作根本不会触发。
• 普通CassandraRepository的save是阻塞同步方法，调用时立即执行写入，所以功能正常。
• 此外，在subscribe内部处理单个元素并调用save，还会导致错误静默丢失、背压处理缺失等问题。

解决方案 #

修改代码以符合Reactive编程范式，确保每个写入操作被正确触发，并处理错误：

优化后的保存方法 #

@Autowired
private AuditLogRepository auditLogRepository;

public Mono<Void> saveAuditLog(Flux<AuditLogMessage> auditLogMessageFlux) {
    return auditLogMessageFlux
            // 将消息转换为AuditLog实体
            .map(auditLogMessage -> AuditLog.builder()
                    .id(UUID.randomUUID().toString())
                    .tenantId(auditLogMessage.getTenantId())
                    .logTime(auditLogMessage.getLogTime())
                    .entity(auditLogMessage.getEntity())
                    .userId(auditLogMessage.getUserId())
                    .logType(auditLogMessage.getLogType())
                    .entityReference(auditLogMessage.getEntityReference())
                    .change(auditLogMessage.getChange())
                    .build())
            // 合并每个save操作的Mono到Flux中，触发写入
            .flatMap(auditLogRepository::save)
            // 添加错误处理，避免异常静默丢失
            .doOnError(error -> System.err.println("保存审计日志失败: " + error.getMessage()))
            // 等待所有操作完成后返回完成信号
            .then();
}

关键改进点 #

1. 使用map进行对象转换，替代原subscribe内的实体构建，更符合Reactive流处理逻辑。
2. 通过flatMap触发每个save操作的执行（save返回的Mono会被订阅）。
3. 添加doOnError处理异常，避免错误被静默忽略。
4. 返回Mono<Void>，让上层调用者（如Kafka消息处理器）管理流的订阅，Spring框架可自动处理上下文和生命周期。

如果需要在方法内部主动触发流执行，也可以在末尾调用.subscribe()，但推荐返回Mono/Flux让上层处理，以获得更好的可控性。

内容的提问来源于stack exchange，提问作者AFzal Khan