Получить планировщик цикла событий webflux

Question

Я использую webflux с netty и jdbc , поэтому я блокирую операцию jdbc следующим образом:

static <T> Mono<T> fromOne(Callable<T> blockingOperation) {
    return Mono.fromCallable(blockingOperation)
        .subscribeOn(jdbcScheduler)
        .publishOn(Schedulers.parallel());
}

Операция блокировкибудет обработан jdbcScheduler , и я хочу, чтобы другой конвейер был обработан планировщиком цикла событий webflux .

Как получить планировщик цикла событий webflux?

Answer 1

Я согласен с @Vikram Rawat. Использование jdbc очень опасно, главным образом потому, что jdbc - это блокирующий интерфейс ввода-вывода, а использование реактивной модели цикла обработки событий очень опасно, поскольку в принципе блокировать весь сервер очень просто.

Однако, даже если это экспериментальное усилие, я предлагаю вам оставаться в курсе проекта R2DBC, который может использовать не блокирующий API для SQL, я использовал его для всплеска, и это очень элегантно.

Я могу предоставить вампример взят из моего домашнего проекта на github на основе sprign boot 2.1 и kotlin:

веб-слой

@Configuration
class ReservationRoutesConfig {

    @Bean
    fun reservationRoutes(@Value("\${baseServer:http://localhost:8080}") baseServer: String,
                          reservationRepository: ReservationRepository) =
            router {
                POST("/reservation") {
                    it.bodyToMono(ReservationRepresentation::class.java)
                            .flatMap { Mono.just(ReservationRepresentation.toDomain(reservationRepresentation = it)) }
                            .flatMap { reservationRepository.save(it).toMono() }
                            .flatMap { ServerResponse.created(URI("$baseServer/reservation/${it.reservationId}")).build() }

                }

                GET("/reservation/{reservationId}") {
                    reservationRepository.findOne(it.pathVariable("reservationId")).toMono()
                            .flatMap { Mono.just(ReservationRepresentation.toRepresentation(it)) }
                            .flatMap { ok().body(BodyInserters.fromObject(it)) }
                }

                DELETE("/reservation/{reservationId}") {
                    reservationRepository.delete(it.pathVariable("reservationId")).toMono()
                            .then(noContent().build())
                }
            }
}

уровень хранилища:

class ReactiveReservationRepository(private val databaseClient: TransactionalDatabaseClient,
                                    private val customerRepository: CustomerRepository) : ReservationRepository {

    override fun findOne(reservationId: String): Publisher<Reservation> =
            databaseClient.inTransaction {
                customerRepository.find(reservationId).toMono()
                        .flatMap { customer ->
                            it.execute().sql("SELECT * FROM reservation WHERE reservation_id=$1")
                                    .bind("$1", reservationId)
                                    .exchange()
                                    .flatMap { sqlRowMap ->
                                        sqlRowMap.extract { t, u ->
                                            Reservation(t.get("reservation_id", String::class.java)!!,
                                                    t.get("restaurant_name", String::class.java)!!,
                                                    customer, t.get("date", LocalDateTime::class.java)!!)
                                        }.one()
                                    }
                        }
            }

    override fun save(reservation: Reservation): Publisher<Reservation> =
            databaseClient.inTransaction {
                customerRepository.save(reservation.reservationId, reservation.customer).toMono()
                        .then(it.execute().sql("INSERT INTO reservation (reservation_id, restaurant_name, date) VALUES ($1, $2, $3)")
                                .bind("$1", reservation.reservationId)
                                .bind("$2", reservation.restaurantName)
                                .bind("$3", reservation.date)
                                .fetch().rowsUpdated())
            }.then(Mono.just(reservation))


    override fun delete(reservationId: String): Publisher<Void> =
            databaseClient.inTransaction {
                customerRepository.delete(reservationId).toMono()
                        .then(it.execute().sql("DELETE FROM reservation WHERE reservation_id = $1")
                                .bind("$1", reservationId)
                                .fetch().rowsUpdated())
            }.then(Mono.empty())

}

Я надеюсь, что это может помочь вам

Answer 2

Я настоятельно рекомендую пересмотреть варианты технологии.Если вы собираетесь использовать jdbc, который все еще блокирует, то вам не следует использовать webflux.Это потому, что webflux будет светить в неблокирующем стеке, но в сочетании с Jdbc он будет действовать как узкое место.Производительность на самом деле снизится.