如何使用 kotlin 协程实现 Spring Flux 的多租户？答案

【问题标题】：How to achieve multi tenancy on spring flux with kotlin coroutine?如何使用 kotlin 协程实现 Spring Flux 的多租户？
【发布时间】：2020-01-13 14:38:39
【问题描述】：

对不起我的英语。

我是 spring 和 kotlin 的新手。
我试图解决的问题是在 kotlin 协程中获取租户值。我做了个简单的例子https://github.com/cardid-zz/spring-multitenant-test

我有 TenantContext 类，它持有传递的租户值

@Component
object TenantContext {
    const val DEFAULT: String = "default"

    private val logger = LoggerFactory.getLogger(javaClass)

    private val currentTenant = InheritableThreadLocal<String?>()

    fun getTenant() : String {
        return currentTenant.get() ?: DEFAULT
    }
    fun set(tenantId : String) {
        currentTenant.set(tenantId)
    }
    fun remove() {
        currentTenant.remove()
    }

    fun asContextElement(): ThreadContextElement<String?> {
        logger.debug("[d] asContextElement ${getTenant()}")
        return currentTenant.asContextElement(getTenant())
    }
}

过滤器中的租户值集

@Component
class TenantFilter (
    private val tenantContext: TenantContext
) : WebFilter {
    private val logger = LoggerFactory.getLogger(javaClass)

    private val tenantHeader = "tenant"

    override fun filter(
        serverWebExchange: ServerWebExchange,
        webFilterChain: WebFilterChain
    ): Mono<Void> {
        val tenant = serverWebExchange.request.headers[tenantHeader]

        if (tenant.isNullOrEmpty()) {
            setTenant(TenantContext.DEFAULT)
        } else {
            setTenant(tenant.first())
        }
        logger.debug("[d] currentThread =  ${Thread.currentThread()}")
        return webFilterChain.filter(serverWebExchange)
    }

    private fun setTenant(tenant: String) {
        try {
            tenantContext.set(tenant)
        } catch (e: Exception) {
            throw RuntimeException()
        }
    }
}

在控制器中我有两个端点

@RestController
class RestController(
    private val service: SomeService
) {
    private val logger = LoggerFactory.getLogger(javaClass)

    @PostMapping("/working")
    suspend fun working(@RequestParam("param") param : Int) : ResponseEntity<*> = coroutineScope(){
        val res = async{ service.doSomething(param)}
        return@coroutineScope ResponseEntity.ok(res)
    }

    @PostMapping("/failed")
    suspend fun failed(@RequestBody body: BodyParam) : ResponseEntity<*> = coroutineScope(){
        logger.debug("[d] ${body.toString()}")
        val res = async { service.doSomething(body.value) }

        return@coroutineScope ResponseEntity.ok(res)
    }
}

其中一个正常工作，一个不正常。区别在于参数，work方法有requestparam，failed通过body获取param。

printf "\nworking go"

curl -i -X POST \
   -H "tenant:properTenant" \
   -H "Content-Type:application/json" \
 'http://localhost:8080/working?param=123'

printf "\nfailed go"

curl -i -X POST \
   -H "tenant:properTenant" \
   -H "Content-Type:application/json" \
   -d \
'{"value":21312}' \
 'http://localhost:8080/failed'

working returns
"param = 123 and tenant was = properTenant"

failed returns
"param = 21312 and tenant was = default"

如何从 /failed 方法中获取租户值？

【问题讨论】：

标签： spring kotlin spring-webflux kotlin-coroutines reactor

【解决方案1】：

我对 Kotlin 不够熟悉，无法举一个具体的例子，但是您不应该在 Flux 应用程序中使用 ThreadLocal 或 InheritableThreadLocal，因为您没有像在传统 Web 应用程序中那样为每个请求提供专用线程。这可能就是您得到意外结果的原因。

考虑使用here 中描述的上下文。

【讨论】：