Spring Reactor Kotlin中R2DBC调用存储过程无返回数据问题

相关代码 #

控制器代码 #

@RestController
class RootController (private val svc: ProcessMsgService) {

    @PostMapping(
        value = ["/"]
        , consumes = [MediaType.APPLICATION_XML_VALUE, MediaType.TEXT_XML_VALUE]
        , produces = [MediaType.APPLICATION_XML_VALUE]
    )
    fun acceptXml (@RequestBody iMsg: String): Mono<ResponseEntity<String>> {
        val oXml = XML {compact() }

        val msg = XML1_0.decodeFromString(MyType.serializer(), iMsg)
        return svc.getRespFromDb (msg).map { oMsg ->
            ResponseEntity.ok()
                        .contentType(MediaType.APPLICATION_XML)
                        .body ( oXml.encodeToString (oMsg))
        }
    }
}

Service代码 #

@Service
class ProcessMsgService (private val databaseClient: DatabaseClient) {

    fun getRespFromDb (msg: MyType): Mono<MyType> {
        return procDoc (msg)
        // 实际代码中有更多分支，此处省略
    }

    @Transactional(readOnly = true)
    fun procDoc (msg: MyType): Mono<MyType> {
        return execDbCall (msg)
                .map { r ->
                    msg.copy (directionAttr = "Rs",   // 请求中为"Rq"
                        respTextAttr = "Success",
                        msgData = msg.msgData.copy ()
                    )  // 此处省略了对execDbCall返回结果的处理
                }
    }

    fun execDbCall (doc: MyType): Mono<DbRespBA> {
        val sql = "BEGIN delme.dumb(in1 => :p_in1, in2 => null, out1 => :p_out1, out2 => :p_out2); END;"
        val pIn1 = "in-1"

        return databaseClient.sql(sql)
            .bind("p_in1", Parameters.`in`(R2dbcType.VARCHAR, pIn1))
            .bind("p_out1", Parameters.out(R2dbcType.VARCHAR))
            .bind("p_out2", Parameters.out(R2dbcType.VARCHAR))
            .map { outParameters, _ ->
                DbRespBA(
                        o1 = outParameters.get("p_out1", String::class.java) ?: "undef"
                        , o2 = outParameters.get("p_out2", String::class.java) ?: "undef"
                        )
            }
            .one() // 将Flux转换为Mono<xxx>
    }

    data class DbRespBA (
        val o1: String,
        val o2: String
    )
}

MyType数据类 #

@Serializable
@XmlSerialName("MyType")
data class MyType(
    @XmlSerialName("direction")
    val directionAttr: String,

    @XmlSerialName("resp_text")
    val respTextAttr: String? = null,

    @XmlSerialName("MsgData")
    val msgData: MsgData   // 实际代码中有定义，此处省略
)

本地测试用procDoc逻辑（可正常返回） #

fun procDoc  (msg: MyType): Mono<MyType> {
    return Mono.just (
        msg.copy (directionAttr = "Rs",
            respTextAttr = "Success",
            msgData = msg.msgData.copy ()
        )
    )
}

排查分析与解决方案 #

1. @Transactional(readOnly = true)注解误用 #

这是最可能的根因：

• Spring R2DBC的事务需要适配反应式模型，普通@Transactional的readOnly=true会限制事务内只能执行读操作，但你的存储过程实际执行了写操作（写入日志表），这会导致事务静默回滚，Mono流被终止，没有数据向下游传递。
• 反应式环境中，@Transactional需要配合R2dbcTransactionManager使用，但即使配置正确，readOnly=true也会阻止写操作。

解决：
直接移除procDoc方法上的@Transactional(readOnly = true)注解；如果确实需要事务，改为不带readOnly的@Transactional，并确保已正确配置R2DBC事务管理器。

2. 数据库输出参数读取异常 #

存储过程执行成功，但输出参数读取时可能抛出异常，导致Mono流终止（反应式流默认会静默处理未捕获的错误，不会向上层抛出）。

排查与解决：

• 确认Oracle存储过程的输出参数名称与代码中的p_out1、p_out2完全匹配（Oracle默认大小写不敏感，但如果存储过程定义时用了双引号，则区分大小写）。
• 在execDbCall中添加错误处理，打印异常信息：

return databaseClient.sql(sql)
    // ... 参数绑定逻辑 ...
    .one()
    .onErrorResume { e ->
        println("DB调用异常: ${e.message}")
        Mono.error(e) // 或返回默认值保证流不中断
    }

3. XML序列化验证 #

虽然本地生成数据时正常，但从DB流程生成的MyType对象可能存在序列化问题。在控制器的map中添加日志，确认对象内容和序列化结果：

return svc.getRespFromDb(msg).map { oMsg ->
    println("生成响应对象: $oMsg")
    val xmlStr = oXml.encodeToString(oMsg)
    println("序列化结果: $xmlStr")
    ResponseEntity.ok()
        .contentType(MediaType.APPLICATION_XML)
        .body(xmlStr)
}

4. 反应式流订阅验证 #

Spring WebFlux会自动订阅控制器返回的Mono，但如果事务导致上下文丢失，可能出现流未被正常订阅的情况。移除事务注解后如果恢复正常，即可验证这一点。

内容的提问来源于stack exchange，提问作者Victor Esin