使用 mongodb-reactive 的反应式 Spring 启动应用程序中的多租户

Question

我们如何使用 Mongodb-reactive 存储库在 spring webflux 中创建多租户应用程序？

我在网络上找不到响应式应用程序的任何完整资源。所有可用资源都用于非反应性应用程序。

更新：

在非反应式应用程序中，我们曾经将上下文数据存储在 ThreadLocal 中，但反应式应用程序无法做到这一点，因为存在线程切换。有一种方法可以将上下文信息存储在 WebFilter 内的反应器上下文中，但我不知道如何在 ReactiveMongoDatabaseFactory 类中获取该数据。

谢谢。

Answer 1

这是我对 Spring WebFlux 的非常粗略的工作解决方案 - 他们已经更新了 ReactiveMongoDatabaseFactory - getMongoDatabase 以返回一个 Mono

创建网络过滤器

public class TenantContextFilter implements WebFilter {

private static final Logger LOGGER = LoggerFactory.getLogger(TenantContextFilter.class);

@Override
public Mono<Void> filter(ServerWebExchange swe, WebFilterChain wfc) {
  ServerHttpRequest request = swe.getRequest();
  HttpHeaders headers = request.getHeaders();
  
  if(headers.getFirst("X-TENANT-ID") == null){
      LOGGER.info(String.format("Missing X-TENANT-ID header"));
      throw new ResponseStatusException(HttpStatus.UNAUTHORIZED);
  }
  
  String tenantId = headers.getFirst("X-TENANT-ID");
  
  LOGGER.info(String.format("Processing request with tenant identifier [%s]", tenantId));
            
  return wfc.filter(swe)
            .contextWrite(TenantContextHolder.setTenantId(tenantId));
    
}    

}

创建类以获取上下文（归功于我发现的某个地方）

    public class TenantContextHolder {

    public static final String TENANT_ID = TenantContextHolder.class.getName() + ".TENANT_ID";

    public static Context setTenantId(String id) {
        return Context.of(TENANT_ID, Mono.just(id));
    }

    public static Mono<String> getTenantId() {
        return Mono.deferContextual(contextView -> {
            if (contextView.hasKey(TENANT_ID)) {
                return contextView.get(TENANT_ID);
            }
            return Mono.empty();
        }
        );
    }

    public static Function<Context, Context> clearContext() {
        return (context) -> context.delete(TENANT_ID);
    }

}

我的 spring 安全设置（允许测试的所有请求）

@EnableWebFluxSecurity
@EnableReactiveMethodSecurity
public class SecurityConfig {

    @Bean
    public SecurityWebFilterChain WebFilterChain(ServerHttpSecurity http) {
        return http
                .formLogin(it -> it.disable())
                .cors(it -> it.disable()) //fix this
                .httpBasic(it -> it.disable())
                .csrf(it -> it.disable())
                .securityContextRepository(NoOpServerSecurityContextRepository.getInstance())
                .authorizeExchange(it -> it.anyExchange().permitAll()) //allow anonymous
                .addFilterAt(new TenantContextFilter(), SecurityWebFiltersOrder.HTTP_BASIC)
                .build();
    }

       }

创建租户 Mongo DB 工厂

我还有一些清理默认设置的工作......

public class MultiTenantMongoDBFactory extends SimpleReactiveMongoDatabaseFactory {

    private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantMongoDBFactory.class);
    private final String defaultDb;

    public MultiTenantMongoDBFactory(MongoClient mongoClient, String databaseName) {
        super(mongoClient, databaseName);
        this.defaultDb = databaseName;
    }

    @Override
    public Mono<MongoDatabase> getMongoDatabase() throws DataAccessException {
        return TenantContextHolder.getTenantId()
                .map(id -> {
                    LOGGER.info(String.format("Database trying to retrieved is [%s]", id));
                    return super.getMongoDatabase(id);
                })
                .flatMap(db -> {
                    return db;
                })
                .log();
    }

}

配置类

@Configuration
@EnableReactiveMongoAuditing
@EnableReactiveMongoRepositories(basePackages = {"com.order.repository"})
class MongoDbConfiguration {
    
    @Bean
    public ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory() {
        return new MultiTenantMongoDBFactory(MongoClients.create("mongodb://user:password@localhost:27017"), "tenant_catalog");
    }

    @Bean
    public ReactiveMongoTemplate reactiveMongoTemplate() {
        ReactiveMongoTemplate template = new ReactiveMongoTemplate(reactiveMongoDatabaseFactory());
        template.setWriteResultChecking(WriteResultChecking.EXCEPTION);

        return template;
    }

}

实体类

@Document(collection = "order")
//getters
//setters

测试

创建两个具有相同集合的 mongo db，将不同的文档放在两个中

在 Postman 中，我刚刚使用“X-TENANT-ID”标头和数据库名称作为值（例如租户 12343 或租户 34383）进行了获取请求，一切顺利！

Answer 2

我能够使用 mangodb 在 Spring Reactive 应用程序中实现多租户。负责实现的主要类有：Custom MongoDbFactory 类、用于捕获租户信息的 WebFilter 类（而不是 Servlet Filter）和用于存储租户信息的 ThreadLocal 类。流程很简单：

1. 在 WebFilter 中从请求中捕获租户相关信息，并将其设置在 ThreadLocal 中。在这里，我使用标头发送租户信息：X-Tenant
2. 实现自定义 MondoDbFactory 类并覆盖 getMongoDatabase() 方法以根据 ThreadLocal 类中可用的当前租户返回数据库。

源代码为：

CurrentTenantHolder.java

package com.jazasoft.demo;

public class CurrentTenantHolder {
    private static final ThreadLocal<String> currentTenant = new InheritableThreadLocal<>();

    public static String get() {
        return currentTenant.get();
    }

    public static void set(String tenant) {
        currentTenant.set(tenant);
    }

    public static String remove() {
        synchronized (currentTenant) {
            String tenant = currentTenant.get();
            currentTenant.remove();
            return tenant;
        }
    }
}

TenantContextWebFilter.java

package com.example.demo;

import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilter;
import org.springframework.web.server.WebFilterChain;
import reactor.core.publisher.Mono;

@Component
public class TenantContextWebFilter implements WebFilter {

    public static final String TENANT_HTTP_HEADER = "X-Tenant";

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        if (request.getHeaders().containsKey(TENANT_HTTP_HEADER)) {
            String tenant = request.getHeaders().getFirst(TENANT_HTTP_HEADER);
            CurrentTenantHolder.set(tenant);
        }
        return chain.filter(exchange).doOnSuccessOrError((Void v, Throwable throwable) -> CurrentTenantHolder.remove());
    }
}

MultiTenantMongoDbFactory.java

package com.example.demo;

import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoDatabase;
import org.springframework.dao.DataAccessException;
import org.springframework.data.mongodb.core.SimpleReactiveMongoDatabaseFactory;


public class MultiTenantMongoDbFactory extends SimpleReactiveMongoDatabaseFactory {
    private final String defaultDatabase;

    public MultiTenantMongoDbFactory(MongoClient mongoClient, String databaseName) {
        super(mongoClient, databaseName);
        this.defaultDatabase = databaseName;
    }


    @Override
    public MongoDatabase getMongoDatabase() throws DataAccessException {
        final String tlName = CurrentTenantHolder.get();
        final String dbToUse = (tlName != null ? tlName : this.defaultDatabase);
        return super.getMongoDatabase(dbToUse);
    }
}

MongoDbConfig.java

package com.example.demo;

import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoClients;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.ReactiveMongoClientFactoryBean;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;

@Configuration
public class MongoDbConfig {

    @Bean
    public ReactiveMongoTemplate reactiveMongoTemplate(MultiTenantMongoDbFactory multiTenantMongoDbFactory) {
        return new ReactiveMongoTemplate(multiTenantMongoDbFactory);
    }

    @Bean
    public MultiTenantMongoDbFactory multiTenantMangoDbFactory(MongoClient mongoClient) {
        return new MultiTenantMongoDbFactory(mongoClient, "test1");
    }

    @Bean
    public ReactiveMongoClientFactoryBean mongoClient() {
        ReactiveMongoClientFactoryBean clientFactory = new ReactiveMongoClientFactoryBean();
        clientFactory.setHost("localhost");
        return clientFactory;
    }
}

更新：

在响应式流中，我们不能再将上下文信息存储在 ThreadLocal 中，因为请求未绑定到单个线程，因此，这不是正确的解决方案。

然而，上下文信息可以像这样存储在WebFilter中的reactor Context中。 chain.filter(exchange).subscriberContext(context -> context.put("tenant", tenant));。问题是如何在ReactiveMongoDatabaseFactory 实现类中获取此上下文信息。