4

我将webflux与netty和jdbc一起使用，所以我用下一种方式包装阻塞 jdbc 操作：

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

阻塞操作将由jdbcScheduler处理，我希望其他管道将由webflux event-loop scheduler 处理。

如何获取 webflux 事件循环调度程序？

4

2 回答 2

4

我强烈建议重新审视技术选项。如果你打算使用 jdbc，它仍然阻塞，那么你不应该使用 webflux。这是因为 webflux 将在非阻塞堆栈中大放异彩，但与 Jdbc 结合将成为瓶颈。性能实际上会下降。

于 2018-10-06T05:08:13.200 回答
0

我同意@Vikram Rawat 的观点，使用 jdbc 非常危险，主要是因为 jdbc 是一个阻塞 IO api，使用事件循环反应模型非常危险，因为基本上它很容易阻塞所有服务器。

但是，即使这是一项实验性工作，我建议您继续关注 R2DBC 项目，该项目能够利用 sql 的无阻塞 API，我将它用于峰值，它非常优雅。

我可以为您提供一个示例，该示例取自我在 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())

}

我希望这可以帮助你

于 2019-01-25T19:32:03.873 回答