java - 在 RxJava 2 中正确使用 Observable.create()(最佳实践)

Question

我目前正在构建一个利用 RxJava 2 和 Firebase 的小型社交媒体风格应用程序。我正在使用 MVP 风格的架构，并且我已经使用一个名为 AuthSource 的接口(interface)抽象出我的 AuthService。

为简单起见，我将在我的服务中使用单一方法:

public class FirebaseAuthService implements AuthSource {

private FirebaseAuth auth;
private FirebaseAuth.AuthStateListener listener;

//initialization code

@Override
public Maybe<User> getUser() {
    return Maybe.create(new MaybeOnSubscribe<User>() {
                            @Override
                            public void subscribe(final MaybeEmitter<User> e) throws Exception {
                                if (auth == null) {
                                    auth = FirebaseAuth.getInstance();
                                }

                                if (listener != null) {
                                    auth.removeAuthStateListener(listener);
                                }

                                listener = new FirebaseAuth.AuthStateListener() {
                                    @Override
                                    public void onAuthStateChanged(@NonNull FirebaseAuth firebaseAuth) {
                                        FirebaseUser firebaseUser = firebaseAuth.getCurrentUser();
                                        auth.removeAuthStateListener(listener);
                                        if (firebaseUser != null) {
                                            User user = new User(
                                                    firebaseUser.getDisplayName(),
                                                    firebaseUser.getEmail());

                                            user.setUserId(firebaseUser.getUid());


                                            Uri photoUrl = firebaseUser.getPhotoUrl();
                                            if (photoUrl != null){
                                                user.setProfilePhotoUrl(photoUrl.toString());
                                            }
                                            e.onSuccess(user);
                                        } else {
                                            e.onComplete();
                                        }
                                    }
                                };

                                auth.addAuthStateListener(listener);
                            }
                        }
    );

}

interface AuthSource {
    Maybe<User> getUser();
//Other methods etc.
}

最后，我将展示处理调用的 Presenter 方法:

//from with a Presenter:
@Override
private void getUserData() {
    disposableSubscriptions.add(
            auth.getUser().subscribeOn(schedulerProvider.io())
                    .observeOn(schedulerProvider.ui())
                    .subscribeWith(
                            new DisposableMaybeObserver<User>() {

                                @Override
                                public void onError(Throwable e) {
                                    view.makeToast(R.string.error_retrieving_data);
                                    view.startDispatchActivity();
                                }

                                @Override
                                public void onComplete() {

                                }

                                @Override
                                public void onSuccess(User user) {
                                    ProfilePagePresenter.this.currentUser = user;
                                    view.setName(user.getName());
                                    view.setEmail(user.getEmail());
                                    if (user.getProfilePhotoUrl().equals("")) {
                                        view.setDefaultProfilePhoto();
                                    } else {
                                        view.setProfilePhotoURI(user.getProfilePhotoUrl());
                                    }

                                    getUserProfileFromDatabase();

                                }
                            }
                    )
    );
}

我意识到问题的主题有点笼统，所以我会尝试从这里缩小范围。只要我使用 Create() 从 Firebase 的 API 成功获取数据，我在上面发布的代码有效。问题是，我对使用 RxJava 2 还很陌生，我不确定垃圾收集和内存泄漏的幕后情况。根据 RxJava 2 文档，我选择使用 Observable.create():

“提供一个 API(通过冷 Observable)，将 react 世界与回调式世界联系起来。”

RxJava 2 Docs最后，我目前为处理这些 Observable 而做的唯一积极主动的事情是，当事件将用户带到新的 Activity 时，在我的 Presenter 中调用 CompositeDisposable.clear() 。

问题:

- 假设在 Presenter 完成时简单地调用 CompositeDisposable.clear() 将处理我的垃圾收集是否安全？ (假设我没有在其余代码中造成内存泄漏)。

-如果我的理解是正确的，在这种情况下，create() 是比 fromCallable() 更好的选择，因为 fromCallable() 应该用于同步事件(即不是类似 Firebase API 回调的东西)？

-它真的像在 Observable.create() 中抛出我的异步回调一样简单吗？我很害怕这样做是多么容易...

Answer 1

Is it safe to assume that simply calling CompositeDisposable.clear() when the Presenter finishes, will handle my Garbage collection? (assuming I haven't created memory leaks in the rest of the code).

它比这更棘手。如果 Observable 引用的所有内容都属于 Activity 范围，则非处置 Observable 不会造成内存泄漏。生产者和消费者都将与 Activity 一起被垃圾回收。如果您引用的资源将在 Activity 中存活下来，例如在 Application 级别实例化的提供程序，则可能会发生内存泄漏。因此，如果您想使用 CompositeDisposable.clear()，请确保在 Observable.create() 中实现 emitter.setCancellable() 来处理那些泄漏资源。

If my understanding is correct, create() is a better option to use than fromCallable() in this case, as fromCallable() should be used for Synchronous events (i.e. not something like Firebase API callbacks)?

create() 用于命名为 fromAsync()。使用 fromCallable() 包装同步方法调用，使用 create() 包装回调代码。

Is it really as simple as just throwing my Asynchronous callbacks in Observable.create()? I'm terrified at how easy that is to do...

这很容易...如果您像第一点提到的那样处理范围之外的那些讨厌的引用。

通常在 Android 上，内存泄漏涉及到很大的 Context。请务必测试您的代码。 leakcanary对这件事有很大的帮助。

最后，您可以通过使用现有的 Firebase RxJava 绑定(bind)来避免自己进行包装。或者从中获取灵感:

• https://github.com/kunny/RxFirebase
• https://github.com/ashdavies/rx-firebase
• https://github.com/DariusL/RxFirebaseAndroid
• https://github.com/ezhome/Android-RxFirebase
• https://github.com/nmoskalenko/RxFirebase
• https://github.com/VictorAlbertos/RxFcm