hibernate - Vertx - 是否可以结合 RxJava 和 Hibernate

Question

我想知道是否可以使用它:

https://github.com/vert-x3/vertx-rx

和

https://github.com/vert-x3/vertx-jdbc-client

并为命名查询/DAO 访问添加 Hibernate。

如何使 Hibernate 非阻塞 I/O？

Answer 1

最近我遇到了完全相同的问题并编写了该库:
https://github.com/eraga/rxjpa2

这是一个如何使用它的示例 vertx .

假设您存储 User您使用 JPA2 Hibernate 访问的 DB 中的实体。那么您将拥有 UsersDAOServiceImpl像那样:

package jpa.rxjava.vertx.sample

import io.vertx.core.AsyncResult
import io.vertx.core.Handler
import io.vertx.reactivex.CompletableHelper
import io.vertx.reactivex.SingleHelper
import data.User // That's your data class
import net.eraga.rxjpa2.*
import javax.persistence.EntityManager
import javax.persistence.EntityManagerFactory

class UsersDAOServiceImpl(
        emf: EntityManagerFactory,
        readyHandler: Handler<AsyncResult<UsersDAOService>>
) : UsersDAOService {

    private lateinit var entityManager: EntityManager

    init {
        emf.rxCreateEntityManager()
                .map {
                    entityManager = it
                    this
                }
                .subscribe(SingleHelper.toObserver(readyHandler))

    }


    override fun fetchByLogin(login: String, resultHandler: Handler<AsyncResult<User>>) {
        entityManager.createQuery("SELECT i From User i where login = :login", User::class.java)
                .setParameter("login", login)
                .rxSingleResult<User>()
                .subscribe(SingleServiceHelper.toObserver(resultHandler))
    }

    override fun fetchAll(resultHandler: Handler<AsyncResult<List<User>>>) {
        entityManager.createQuery("SELECT i From User i order by modified desc", User::class.java)
                .rxResultList()
                .subscribe(SingleServiceHelper.toObserver(resultHandler))
    }

    override fun fetchById(id: Int, resultHandler: Handler<AsyncResult<User>>) {
        entityManager.rxFind(User::class.java, id)
                .subscribe(SingleServiceHelper.toObserver(resultHandler))
    }

    override fun create(user: User, resultHandler: Handler<AsyncResult<Void>>) {
        entityManager
                .rxPersist(user)
                .subscribe(CompletableHelper.toObserver(resultHandler))
    }

    override fun save(user: User, resultHandler: Handler<AsyncResult<Void>>) {
        entityManager
                .rxMerge(user)
                .toCompletable()
                .subscribe(CompletableHelper.toObserver(resultHandler))
    }

    override fun delete(id: Int, resultHandler: Handler<AsyncResult<Void>>) {
        val user = User().apply { this.id = id }
        entityManager
                .rxRemove(user)
                .subscribe(CompletableHelper.toObserver(resultHandler))
    }
}

通知 SingleServiceHelper帮助翻译的类(class) Single转至 SingleObserver s:

package jpa.rxjava.vertx.sample

import com.fasterxml.jackson.core.type.TypeReference
import io.reactivex.Single
import io.reactivex.SingleObserver
import io.reactivex.SingleTransformer
import io.reactivex.annotations.NonNull
import io.reactivex.disposables.Disposable
import io.vertx.core.AsyncResult
import io.vertx.core.Future
import io.vertx.core.Handler
import io.vertx.core.buffer.Buffer
import io.vertx.reactivex.core.json.SingleUnmarshaller
import io.vertx.serviceproxy.ServiceException
import java.util.concurrent.atomic.AtomicBoolean
import java.util.function.Function


object SingleServiceHelper {
    /**
     * Adapts an Vert.x `Handler<AsyncResult<T>>` to an RxJava2 [SingleObserver].
     *
     *
     * The returned observer can be subscribed to an [Single.subscribe].
     *
     * @param handler the handler to adapt
     * @return the observer
     */
    fun <T> toObserver(handler: Handler<AsyncResult<T>>): SingleObserver<T> {
        val completed = AtomicBoolean()
        return object : SingleObserver<T> {
            override fun onSubscribe(@NonNull d: Disposable) {}
            override fun onSuccess(@NonNull item: T) {
                if (completed.compareAndSet(false, true)) {
                    handler.handle(Future.succeededFuture(item))
                }
            }

            override fun onError(error: Throwable) {
                if (completed.compareAndSet(false, true)) {
                    val ex = ServiceException(-1,error.message).apply {
                        initCause(error)
                    }
                    handler.handle(Future.failedFuture(ex))
                }
            }
        }
    }

    fun <T> unmarshaller(mappedType: Class<T>): SingleTransformer<Buffer, T> {
        return SingleUnmarshaller(Function.identity(), mappedType)
    }

    fun <T> unmarshaller(mappedTypeRef: TypeReference<T>): SingleTransformer<Buffer, T> {
        return SingleUnmarshaller(Function.identity(), mappedTypeRef)
    }
}

这就是你如何实例化 EntityManagerFactory使用 RxPersistence当您启动 App verticle 时:

package jpa.rxjava.vertx.sample

import io.reactivex.plugins.RxJavaPlugins
import io.vertx.core.DeploymentOptions
import io.vertx.core.Future
import io.vertx.core.VertxOptions
import io.vertx.core.json.JsonObject
import io.vertx.reactivex.core.AbstractVerticle
import io.vertx.reactivex.core.RxHelper
import io.vertx.reactivex.core.Vertx
import net.eraga.rxjpa2.RxPersistence
import org.slf4j.LoggerFactory
import javax.persistence.EntityManagerFactory

class App : AbstractVerticle() {
    private val log = LoggerFactory.getLogger(this.javaClass)

    @Throws(Exception::class)
    override fun start(startFuture: Future<Void>) {

        RxPersistence
                .createEntityManagerFactory("initiatives")
                .flatMap {
                    log.info("Done createEntityManagerFactory {}", it)
                    App.entityManagerFactory = it
                    vertx.rxDeployVerticle(UsersVerticle::class.java.name)
                }.flatMap {
                    log.info("Started {}", it)

                    vertx.rxDeployVerticle(HttpServerVerticle::class.java.name,
                            DeploymentOptions().setInstances(2))
                }.subscribe(
                        {
                            log.info("Started {}", it)
                            startFuture.complete()
                        },
                        {
                            log.error("Fail {}", it.message)
                            startFuture.fail(it)
                        }
                )
    }

    companion object {
        lateinit var entityManagerFactory: EntityManagerFactory

        @JvmStatic
        fun main(args: Array<String>) {
            System.setProperty("org.jboss.logging.provider", "slf4j")
            System.setProperty("hazelcast.logging.type", "slf4j")


            Vertx.clusteredVertx(VertxOptions().setClustered(true)) { cluster ->
                if (cluster.succeeded()) {
                    val vertx = cluster.result()

                    RxJavaPlugins.setComputationSchedulerHandler({ s -> RxHelper.scheduler(vertx) })
                    RxJavaPlugins.setIoSchedulerHandler({ s -> RxHelper.blockingScheduler(vertx) })
                    RxJavaPlugins.setNewThreadSchedulerHandler({ s -> RxHelper.scheduler(vertx) })
                    RxJavaPlugins.lockdown()

                    vertx.deployVerticle(App::class.java.name, DeploymentOptions().setConfig(JsonObject().put("local", true)))
                }
            }
        }
    }
}