RxAndroid - 修复活动泄漏？

Question

我已将内容提炼为下面显示的 Activity，以及关联的 build.config 和 activity_main.xml。

问题是，如果您在计数器完成之前按 BACK，LeakCanary 会报告 Activity 泄漏，而让它运行到最后不会。

我知道发生了什么。我知道必须有一个由观察者在 Activity 中创建的匿名内部类，它在 onPause() 调用中仍然存在，因为如果您观看控制台，您可以看到它继续计数。我知道此时我已经取消订阅，但进程中的线程仍然作为 Activity 的内部类运行，这是泄漏的标志。这对我来说也不仅仅是一个人为的极端情况，在我的真实应用程序中，计数应该在方向改变后继续，所以我坚持保留片段中必要的内容并相应地获取计数，但泄漏相同。我只是不需要在这里展示它来简化示例。

MainAcivity.java：

package com.otamate.rxactivityleaker;

import android.os.Bundle;
import android.os.SystemClock;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.TextView;

import com.squareup.leakcanary.LeakCanary;

import rx.Observable;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity {
    private Observable<Long> mObservable;
    private Subscriber<Long> mSubscriber;

    public TextView mTextView;
    public final static int MAX_PROGRESS = 10;
    public final static int EMIT_DELAY_MS = 1000;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        LeakCanary.install(getApplication());

        setContentView(R.layout.activity_main);

        mTextView = (TextView) findViewById(R.id.textView);

        mSubscriber = createSubscriber();
        mObservable = createObservable();

        mObservable.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(mSubscriber);
    }

    private Observable createObservable() {
        return Observable.create (
            new Observable.OnSubscribe<Long>() {

                @Override
                public void call(Subscriber<? super Long> sub) {
                    for (long i = 1; i < MAX_PROGRESS + 1; i++) {

                        Log.d("Observable", "Progress: " + i);

                        sub.onNext(i);
                        SystemClock.sleep(EMIT_DELAY_MS);
                    }
                    sub.onCompleted();
                }
            }
        );
    }

    private Subscriber createSubscriber() {
        return new Subscriber<Long>() {

            @Override
            public void onNext(Long val) {
                Log.d("Subscriber", "Loop " + val);

                mTextView.setText("Progress: " + val);
            }

            @Override
            public void onCompleted() {
                Log.d("Subscriber", "Completed");

                mTextView.setText("Done!");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("Subscriber", "Error: " + e);

                mTextView.setText("Error!");
            }
        };
    }

    @Override
    public void onPause() {
        super.onPause();

        if (mSubscriber != null) {
            Log.d("MainActivity", "onPause() Unsubscribed");

            mSubscriber.unsubscribe();
        }
    }
}

这里是activity_main.xml：

<?xml version="1.0" encoding="utf-8"?>
<RelativeLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    android:paddingBottom="@dimen/activity_vertical_margin"
    android:paddingLeft="@dimen/activity_horizontal_margin"
    android:paddingRight="@dimen/activity_horizontal_margin"
    android:paddingTop="@dimen/activity_vertical_margin"
    tools:context="com.otamate.rxactivityleaker.MainActivity">

    <TextView
        android:id="@+id/textView"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:text="Idle" />
</RelativeLayout>

build.gradle：

apply plugin: 'com.android.application'

android {
    compileSdkVersion 23
    buildToolsVersion "23.0.2"

    defaultConfig {
        applicationId "com.otamate.rxactivityleaker"
        minSdkVersion 15
        targetSdkVersion 23
        versionCode 1
        versionName "1.0"
    }
    buildTypes {
        release {
            minifyEnabled false
            proguardFiles getDefaultProguardFile('proguard-android.txt'), 'proguard-rules.pro'
        }
    }
}

dependencies {
    compile fileTree(dir: 'libs', include: ['*.jar'])
    testCompile 'junit:junit:4.12'
    compile 'com.android.support:appcompat-v7:23.1.1'
    compile 'com.android.support:appcompat-v7:23.1.1'
    compile 'io.reactivex:rxandroid:0.25.0'
    debugCompile 'com.squareup.leakcanary:leakcanary-android:1.3.1'
    releaseCompile 'com.squareup.leakcanary:leakcanary-android-no-op:1.3.1'
}

Answer 1

Observable.create 变成easy for you to shoot yourself in the foot。

你的问题在这里：

private Observable createObservable() {
    return Observable.create (
        new Observable.OnSubscribe<Long>() {

            @Override
            public void call(Subscriber<? super Long> sub) {
                for (long i = 1; i < MAX_PROGRESS + 1; i++) {

                    Log.d("Observable", "Progress: " + i);

                    sub.onNext(i);
                    SystemClock.sleep(EMIT_DELAY_MS);
                }
                sub.onCompleted();
            }
        }
    );
}

在创建 Observable 时，您不会在调用 onNext/onCompleted 之前检查 Subscriber 是否取消订阅。你也没有处理背压。

查看AbstractOnSubscribe，它可以帮助您更轻松地创建Observable。

编辑：AbstractOnSubscribe 在 1.1 中被删除，但也许 SyncOnSubscribe 会起作用。

Answer 2

试试这个

Subscription subscription;
@Override 
protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);

    LeakCanary.install(getApplication()); 

    setContentView(R.layout.activity_main);

    mTextView = (TextView) findViewById(R.id.textView);

    mSubscriber = createSubscriber();
    mObservable = createObservable(); 

    subscription = mObservable.subscribeOn(Schedulers.io()) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(mSubscriber);
} 

比在 onPause() 中

@Override 
public void onPause() { 
    super.onPause(); 

    if (subscription != null) {

        subscription.unsubscribe();
    } 
}