如何使用Spring R2DBC实现jdbcTemplate.queryForStream等效的简单流式查询？

核心工具：DatabaseClient #

Spring R2DBC提供的DatabaseClient是JdbcTemplate的非阻塞等效工具，完全不需要实体映射或Spring Data R2DBC的Repository层，适合执行基础非阻塞查询。

第一步：添加依赖 #

如果用Maven，添加Spring R2DBC starter和对应数据库驱动（以PostgreSQL为例）：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-postgresql</artifactId>
    <scope>runtime</scope>
</dependency>

Gradle配置：

implementation 'org.springframework.boot:spring-boot-starter-data-r2dbc'
runtimeOnly 'io.r2dbc:r2dbc-postgresql'

第二步：配置数据库连接 #

在application.yml中配置R2DBC连接信息：

spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/your_db
    username: your_user
    password: your_password

第三步：执行流式查询（Java示例） #

直接注入DatabaseClient，执行查询并处理流式结果，等价于queryForStream：

import org.springframework.r2dbc.core.DatabaseClient;
import reactor.core.publisher.Flux;

@Component
public class RawQueryService {

    private final DatabaseClient databaseClient;

    public RawQueryService(DatabaseClient databaseClient) {
        this.databaseClient = databaseClient;
    }

    // 流式查询，类似jdbcTemplate.queryForStream
    public Flux<String> fetchUsernames() {
        String sql = "SELECT username FROM users WHERE active = true";
        
        return databaseClient.sql(sql)
                .map(row -> row.get("username", String.class)) // 手动映射行数据，无需实体类
                .all(); // 返回Flux，代表非阻塞流式结果
    }

    // 带参数的流式查询
    public Flux<String> fetchUsernamesByRole(String role) {
        String sql = "SELECT username FROM users WHERE role = :role";
        
        return databaseClient.sql(sql)
                .bind("role", role)
                .map(row -> row.get("username", String.class))
                .all();
    }
}

Flux对应Stream的非阻塞响应式版本，可通过subscribe()或结合Spring WebFlux消费流数据。

第四步：Kotlin协程适配（推荐） #

借助Spring协程支持，将响应式Flux转为协程Flow，用法更贴合Kotlin习惯：

先添加协程依赖：

implementation 'org.springframework.boot:spring-boot-starter-webflux'
implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-reactor'

编写协程版本查询：

import org.springframework.r2dbc.core.DatabaseClient
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.reactive.asFlow

@Component
class RawQueryService(
    private val databaseClient: DatabaseClient
) {

    // 协程Flow版本，等价于queryForStream
    suspend fun fetchUsernames(): Flow<String> {
        val sql = "SELECT username FROM users WHERE active = true"
        
        return databaseClient.sql(sql)
            .map { row -> row.get("username", String::class.java) }
            .all()
            .asFlow() // 将Flux转为协程Flow
    }

    // 带参数的协程查询
    suspend fun fetchUsernamesByRole(role: String): Flow<String> {
        val sql = "SELECT username FROM users WHERE role = :role"
        
        return databaseClient.sql(sql)
            .bind("role", role)
            .map { row -> row.get("username", String::class.java) }
            .all()
            .asFlow()
    }
}

可以直接在协程函数中用for循环消费Flow，体验和CoroutineCrudRepository一致。

关键说明 #

• DatabaseClient.all()返回的Flux不会一次性加载所有数据到内存，和queryForStream的流式特性一致。
• 手动映射行数据，完全无需实体类或JPA风格的映射逻辑，满足简单查询需求。
• 协程版本通过asFlow()转换，实现非阻塞且符合Kotlin编程习惯的查询方式。

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