与 rxjava Android 同步调用

Question

我的应用有一个 SearchView。当用户在 SearchView 中键入时，onQueryTextChange 会将查询传递给演示者，然后它调用 API。我正在使用 Retrofit 和 RxJava 进行调用。这些调用返回一个 json 文件，其中包含用户迄今为止键入的内容。问题是，如果用户快速输入字母并且网络速度很慢，有时 SearchView 不会根据所有输入的字母显示结果，但可能会显示到倒数第二个，因为最后一次调用获得结果的速度更快与倒数第二个相比。

示例： 用户开始输入：

"cou" -> 调用 API（3 个字母后第一次调用）-> 开始返回值

"n" -> 拨打电话 -> 开始返回值

"t" -> 拨打电话 -> 开始返回值

"r" -> 拨打电话（连接慢）

"y" -> 拨打电话 -> 开始返回值

-> "r" 最终得到结果并返回它们

public Observable<List<MyModel>> getValues(String query) {
    return Observable.defer(() -> mNetworkService.getAPI()
            .getValues(query)
            .retry(2)
            .onErrorReturn(e -> new ArrayList<>()));
}

调用非常简单，每当我遇到错误时，我都不想显示任何内容。

有没有办法解决这个问题？或者也许这不是使用反应式编程的情况？

编辑： 为了更清楚，流程如下：

1. 使用自定义搜索视图的活动 (https://github.com/Mauker1/MaterialSearchView)

2. 当用户开始输入时，自定义搜索视图有一个监听器。一旦用户开始输入，Activity 就会调用 Presenter。

3. presenter 将订阅一个由交互器返回的 observable：

主持人：

addSubscription(mInteractor.getValues(query)
            .observeOn(mMainScheduler)
            .subscribeOn(mIoScheduler)
            .subscribe(data -> {
                getMvpView().showValues(data);
            }, e -> {
                Log.e(TAG, e.getMessage());
            }));

交互者：

public Observable<List<MyModel>> getValues(String query) {
    return Observable.defer(() -> mNetworkService.getAPI()
            .getValues(query)
            .debounce(2, TimeUnit.SECONDS)
            .retry(2)
            .onErrorReturn(e -> new ArrayList<>()));

所以现在我要么在“正常”搜索视图中更改自定义搜索视图，然后使用 RxBinding，要么我应该使用处理程序或类似的东西（但仍在努力如何将其适应我的架构）

Answer 1

首先将您的 Searchview 设为 Observable，以便您可以应用 Rx 运算符。 将 searchview 转换为 Observable

public static Observable<String> fromview(SearchView searchView) {
final PublishSubject<String> subject = PublishSubject.create();

searchView.setOnQueryTextListener(new SearchView.OnQueryTextListener() {
    @Override
    public boolean onQueryTextSubmit(String s) {
        subject.onComplete();
        searchView.clearFocus(); //if you want to close keyboard
        return false;
    }

    @Override
    public boolean onQueryTextChange(String text) {
        subject.onNext(text);
        return false;
    }
});

return subject;

}

private void observeSearchView() {

disposable = RxSearchObservable.fromview(binding.svTweet)
        .debounce(300, TimeUnit.MILLISECONDS)
        .filter(text -> !text.isEmpty() && text.length() >= 3)
        .map(text -> text.toLowerCase().trim())
        .distinctUntilChanged()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe()

}

您可以应用过滤器、条件 RxJava debounce() 运算符延迟执行任何操作，直到用户短暂暂停。

distinctUntilChanged() 的使用确保用户可以搜索相同的东西两次，但不能立即背靠背

filter运算符用于过滤不需要的字符串，如本例中的空字符串，以避免不必要的网络调用。

Handling searchview withRXJava

Answer 2

你很幸运，有一个叫做 debounce 的运算符

Observable.defer(() -> mNetworkService.getAPI()
            .getValues(query)
            .debounce(3, TimeUnit.SECONDS)
            .retry(2)
            .onErrorReturn(e -> new ArrayList<>()));

debounce 的作用是在继续之前等待 N 个时间单位以获得更多结果。例如，网络需要 2 秒才能返回，并且您在请求后用请求淹没它，去抖动将等待 3 秒没有结果，然后返回最后一个结果。可以将其视为放弃除 N 次不活动之前的所有内容。

这解决了您的问题，但仍会淹没网络，理想情况下，您将使用优秀的 RxBinding 库在发出请求之前执行延迟，例如：

RxTextView.textChanges(searchView)
.debounce(3, TimeUnit.SECONDS)
.map(input->mNetworkService.getAPI().getValues(input.queryText().toString()))
.retry(2)
.onErrorReturn(e -> new ArrayList<>()))

使用当前设置，它会在用户输入内容后等待 3 秒，然后才进行网络调用。相反，如果他们开始输入新内容，则第一个待处理的搜索请求将被丢弃。

编辑：基于 OP 更改为 RxTextView.textChanges(textview) 不使用 android SearchView 小部件

Answer 3

扩展@MikeN 所说的内容，如果您只想使用 LAST 输入的结果，您应该使用 switchMap()（在其他一些 Rx 实现中是 flatMapLatest()）。

Answer 4

我在不使用 RxBinding 的情况下解决了泛滥问题，我想发布我的解决方案以防其他人需要它。 因此，每当调用 onTextChanged 时，我首先检查大小是否 > 2 以及它是否连接到网络（由 BroadcastReceiver 更新的布尔值）。然后我创建要发送的消息已延迟，并删除队列中的所有其他消息。这意味着我将只执行不在指定延迟内的查询：

@Override
    public void onTextChanged(CharSequence s, int start, int before, int count) {

        if (TextUtils.getTrimmedLength(s) > 2 && isConnected) {
            mHandler.removeMessages(QUERY_MESSAGE);
            Message message = Message.obtain(mHandler, QUERY_MESSAGE, s.toString().trim());
            mHandler.sendMessageDelayed(message, MESSAGE_DELAY_MILLIS);
        }
    }

然后是处理程序：

private Handler mHandler = new Handler() {
    @Override
    public void handleMessage(Message msg) {
        if (msg.what == QUERY_MESSAGE) {
            String query = (String)msg.obj;
            mPresenter.getValues(query);
        }
    }
};

Answer 5

1. 将 rxbinding 依赖添加到 gradle 实现 "com.jakewharton.rxbinding2:rxbinding-kotlin:2.1.1"
2. 使用 debounce 和 distinct 忽略频繁的按键输入和重复输入
3. 释放之前的 API 调用以仅获取最新的搜索结果

    override fun onCreateOptionsMenu(menu: Menu, inflater: MenuInflater) {
        inflater.inflate(R.menu.toolbar_menu, menu)

        // Associate searchable configuration with the SearchView
        val searchManager = requireContext().getSystemService(Context.SEARCH_SERVICE) as SearchManager
        searchView = menu.findItem(R.id.action_search).actionView as SearchView
        searchView.setSearchableInfo(
            searchManager.getSearchableInfo(requireActivity().componentName)
        )
        searchView.maxWidth = Integer.MAX_VALUE

        // listening to search query text change
        disposable = RxSearchView.queryTextChangeEvents(searchView)
            .debounce(750, TimeUnit.MILLISECONDS)
            .distinctUntilChanged()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                callApi(it.queryText().toString())
            }, {
                Timber.e(it)
            })

        super.onCreateOptionsMenu(menu, inflater)
    }

    private fun callApi(query: String){
        if(!apiDisposable.isDisposed){
            apiDisposable.dispose()
        }
        apiDisposable = mNetworkService.getAPI(query)
    }