为什么使用mono.FromCallable（）时会出现“阻塞，线程反应器-http-nio-4不支持”的错误？

Question

我同意反应性代码不应该阻塞，但是...

在下面的方法findTenantStorageFactory（）中，我需要阻止tenantId，这样该方法就可以为租户返回一个特定的工厂。我从Project Reactor中找到了一些说明，说明我需要将方法包装在一个mono.FromCallable（）中，并使用boundedElastic（）调度程序。但是，当我在调试器中运行此代码时，我会收到可怕的：“block（）/blockfirst（）/blocklast（）正在阻塞，这是不受支持的”错误。我尝试了无数个想法，但没有任何进展。我很感激你的建议。

//////////////////////////////////////////////////////////////////////////
fun findTenantStorageFactory(tenantId: Mono<TenantId>):
  Mono<Either<BaseAzureBlobStorageException, MultitenantAzureBlobStorageFactory>> {
    val myValue: Callable<Either<
                   BaseAzureBlobStorageException, 
                   MultitenantAzureBlobStorageFactory>> =

    Callable { lookupTenantBlobStorageFactory(tenantId.block()!!, factories) }

    return Mono.fromCallable<Either<
                  BaseAzureBlobStorageException, 
                  MultitenantAzureBlobStorageFactory>>(myValue)
               .subscribeOn(Schedulers.boundedElastic())
    }

//////////////////////////////////////////////////////////////////////////
private val blobStorageClientBuilder: AzureBlobClientBuilder
 get() = 
   findTenantStorageFactory(tenantId).block()!!
     .fold({ throw it }, { it.blobStorageClient })!!

/////////////////////////////// ERROR ///////////////////////////////////
    java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-4
    at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:83) ~[reactor-core-3.4.0-M1.jar:3.4.0-M1]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 

Answer 1

为什么您认为需要阻止tenantid？我将使用map将单声道转换为单声道。如果LookupTenantBlobStorageFactory是一个长期运行的方法，并且您希望避免阻塞当前线程，那么Publishon应该会对您有所帮助。类似于这样：

fun findTenantStorageFactory(tenantId: Mono<TenantId>):
  Mono<Either<BaseAzureBlobStorageException, MultitenantAzureBlobStorageFactory>> {
     return tenantId.publishOn(Schedulers.boundedElastic())
        .map({ id -> lookupTenantBlobStorageFactory(id, factories) })
}