小编Sam*_*iel的帖子

我是新手使用rxjava并且我正在尝试使用rxjava2在后台运行一个函数,但该方法未被调用我正在使用的代码如下所示让我知道它是否是在后台执行函数的正确方法:

Observable.fromCallable<OrderItem>(Callable {
        saveToDb(existingQty, newOty, product_id)
    }).doOnSubscribe {
        object : Observable<OrderItem>() {
            override fun subscribeActual(emitter: Observer<in OrderItem>?) {
                try {
                    val orderItem = saveToDb(existingQty, newOty, product_id)
                    emitter?.onNext(orderItem)
                    emitter?.onComplete()
                } catch (e: Exception) {
                    emitter?.onError(e)
                }
            }
        }
    }
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread()).doOnSubscribe {
        object : Observer<OrderItem> {
            override fun onComplete() {
            }

            override fun onNext(t: OrderItem) {

            }

            override fun onError(e: Throwable) {
            }

            override fun onSubscribe(d: Disposable) {
            }
        }
    }