基于Schema的Spring Webflux+R2DBC多租户应用实现求助

我来帮你梳理下针对这个场景的可行方案，结合你提到的几个思路，重点说说怎么用连接池+动态设置Schema的方式来实现——这应该是最符合你需求、兼顾效率和扩展性的方向：

核心思路 #

PostgreSQL支持在连接级别动态切换Schema，我们可以复用r2dbc-pool连接池，在连接被借出给请求时，根据当前租户上下文动态设置对应的Schema，连接归还时再重置为默认Schema，避免租户间的Schema污染。这和Hibernate的MultiTenantConnectionProvider思路本质是一致的。

1. 实现租户上下文管理（适配响应式场景） #

Webflux是响应式编程模型，不能用ThreadLocal存储租户信息，我们需要基于Reactor的Context来传递租户ID：

public class TenantContext {
    public static final String TENANT_ID_KEY = "tenantId";

    // 从Reactor Context中获取当前租户ID
    public static Mono<String> getCurrentTenantId() {
        return Mono.deferContextual(contextView -> 
            Mono.justOrEmpty(contextView.getOrDefault(TENANT_ID_KEY, null))
        );
    }

    // 往Reactor Context中设置租户ID的工具方法
    public static <T> Function<Mono<T>, Mono<T>> withTenantId(String tenantId) {
        return mono -> mono.contextWrite(context -> context.put(TENANT_ID_KEY, tenantId));
    }
}

2. 自定义租户感知的PostgreSQL连接工厂 #

重写PostgresqlConnectionFactory的prepareConnection方法，从Reactor Context中获取租户ID并动态设置Schema：

public class TenantAwarePostgresqlConnectionFactory extends PostgresqlConnectionFactory {

    public TenantAwarePostgresqlConnectionFactory(PostgresqlConnectionConfiguration configuration) {
        super(configuration);
    }

    @Override
    protected Mono<Void> prepareConnection(PostgresqlConnection connection) {
        return TenantContext.getCurrentTenantId()
                .flatMap(tenantId -> {
                    if (tenantId == null) {
                        // 无租户ID时使用默认逻辑（比如连接默认Schema）
                        return super.prepareConnection(connection);
                    }
                    // 使用参数化SQL避免注入风险，替换字符串拼接
                    return connection.createStatement("SET SCHEMA $1")
                            .bind(0, tenantId)
                            .execute()
                            .then();
                })
                .switchIfEmpty(super.prepareConnection(connection));
    }
}

3. 配置R2DBC连接池 #

把自定义的连接工厂作为底层工厂接入r2dbc-pool，同时添加连接归还时的Schema重置逻辑，避免租户信息残留：

@Configuration
public class R2dbcConfig {

    @Value("${spring.r2dbc.url}")
    private String url;

    @Value("${spring.r2dbc.username}")
    private String username;

    @Value("${spring.r2dbc.password}")
    private String password;

    @Bean
    public ConnectionFactory connectionFactory() {
        // 基础PostgreSQL配置，不指定固定Schema
        PostgresqlConnectionConfiguration config = PostgresqlConnectionConfiguration.builder()
                .url(url)
                .username(username)
                .password(password)
                .build();

        TenantAwarePostgresqlConnectionFactory tenantAwareFactory = new TenantAwarePostgresqlConnectionFactory(config);

        // 配置连接池，添加归还时的Schema重置
        return ConnectionPool.builder(tenantAwareFactory)
                .maxIdleTime(Duration.ofMinutes(30))
                .maxSize(10)
                .connectionDecorator(connection -> 
                        // 归还连接时重置为默认Schema（比如public）
                        Mono.from(connection.createStatement("SET SCHEMA public")
                                .execute()
                                .then())
                                .thenReturn(connection)
                )
                .build();
    }

    // 配置DatabaseClient（Spring Data R2DBC会自动复用这个ConnectionFactory）
    @Bean
    public DatabaseClient databaseClient(ConnectionFactory connectionFactory) {
        return DatabaseClient.create(connectionFactory);
    }
}

4. 请求入口注入租户上下文 #

编写WebFilter，从请求中提取租户ID（比如从请求头、路径参数或登录用户信息中），并放入Reactor Context：

@Component
public class TenantContextFilter implements WebFilter {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        // 这里根据你的业务逻辑获取租户ID，示例从请求头获取
        String tenantId = exchange.getRequest().getHeaders().getFirst("X-Tenant-ID");
        
        if (tenantId == null) {
            // 处理无租户ID的情况，比如返回400错误
            exchange.getResponse().setStatusCode(HttpStatus.BAD_REQUEST);
            return exchange.getResponse().setComplete();
        }

        // 将租户ID写入Reactor Context，后续所有操作都能获取到
        return chain.filter(exchange)
                .contextWrite(context -> context.put(TenantContext.TENANT_ID_KEY, tenantId));
    }
}

5. 验证Spring Data R2DBC兼容性 #

完成上述配置后，你的Spring Data R2DBC Repository可以完全正常使用，不需要修改任何SQL或Repository代码：

public interface CityRepository extends ReactiveCrudRepository<City, Long> {
    // 自动使用当前租户的Schema查询
    Flux<City> findByName(String name);
}

关键注意事项 #

• Schema存在性校验：建议在应用启动或首次访问租户时，校验对应Schema是否存在，避免因Schema不存在导致的SQL错误。
• 安全防护：确保租户ID的来源是可信的，禁止用户传入未授权的租户ID，同时始终使用参数化SQL设置Schema，避免注入风险。
• 连接池参数调优：根据业务并发量调整连接池的maxSize、maxIdleTime等参数，平衡资源占用和响应速度。

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