如何用kotlin coroutine在spring flux上实现多租户？

Question

对不起，我的英语不好。

我是个新手，不熟悉spring和kotlin。我想解决的问题是在kotlin coroutine中获取租户值。我做了一个简单的例子 https:/github.comcardid-zzspring-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)
    }
}

其中一个正常工作，一个不正常。不同的是参数，工作的方法有requestparam，失败的方法通过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"                                                                                                                                        

如何从失败的方法中获取租户值？

Answer 1

我对Kotlin还不够熟悉，无法给出具体的例子，但你不应该在flux应用中使用ThreadLocal或InheritableThreadLocal，因为你不像传统的Web应用那样每个请求都有一个专门的线程。这很可能是你得到意外结果的原因。

请考虑使用如下所述的Context 此处.