Webflux a BlockHound

BlockHound je java agent (software nahraný JVM před zavoláním main metody), který způsobí to, že pokud v rámci neblokujícího vlákna (vlákno, které nemá volat blokující volání) zavoláte metodu, která blokuje vlákno, vyhodí se chyba.

Závislost

implementation("io.projectreactor.tools:blockhound:1.0.4.RELEASE")

Pokud si chceme BlockHound přizpůsobit, vytvoříme třídu, která implementuje BlockHoundIntegration

import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Component
import reactor.blockhound.BlockHound
import reactor.blockhound.BlockingOperationError
import javax.annotation.PostConstruct
import reactor.blockhound.integration.BlockHoundIntegration

@Component
class BlockHoundIntegration : BlockHoundIntegration {
    private val logger: Logger = LoggerFactory.getLogger(javaClass)

    @PostConstruct
    fun init() {
        BlockHound.install(this)
    }

    override fun applyTo(builder: BlockHound.Builder) {
        builder.blockingMethodCallback {
            logger.error("Blocking Call!", BlockingOperationError(it))
        }
    }
}

Když nyní zavoláme některý z těchto endpointů

@GetMapping("/blockV1")
fun blockV1(): Mono {
    return Mono.just("Ok")
        .map {
            Thread.sleep(1000L)
            it
        }
}

@GetMapping("/blockV2")
fun blockV2(): Mono {
    logger.info("GET /block called")
    Thread.sleep(100L)
    return Mono.just("Ok")
}

v logu budeme mít chybu

reactor.blockhound.BlockingOperationError: Blocking call! java.lang.Thread.sleep
	at cz.vitfo.webfluxpostgres.blockhound.BlockHoundIntegration$applyTo$1.accept(BlockHoundIntegration.kt:22) ~[main/:na]
	at cz.vitfo.webfluxpostgres.blockhound.BlockHoundIntegration$applyTo$1.accept(BlockHoundIntegration.kt:12) ~[main/:na]
	at reactor.blockhound.BlockHound$Builder.lambda$install$8(BlockHound.java:383) ~[blockhound-1.0.4.RELEASE.jar:na]
	at reactor.blockhound.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:89) ~[na:na]
	at java.base/java.lang.Thread.sleep(Thread.java) ~[na:na]

Pokud chceme aby, program chybu nejen zalogoval, ale aby se propagovala výše, stačí upravit metodu applyTo BlockHoundIntegration

override fun applyTo(builder: BlockHound.Builder) {
    builder.blockingMethodCallback {
        val error = BlockingOperationError(it)
        logger.error("Blocking Call!", error)
        throw error
    }
}

Nyní volání endpointu vrátí

{
    "timestamp": "2021-03-30T07:25:36.233+0000",
    "path": "/blockV1",
    "status": 500,
    "error": "Internal Server Error",
    "message": "Blocking call! java.lang.Thread.sleep"
}

Zdroj: domenicosibilio.medium.com/blockhound-detect-blocking-calls-in-reactive-code-before-its-too-late