带有 IntentService 的 Android RxJava 间隔答案

【问题标题】：Android RxJava Interval with IntentService带有 IntentService 的 Android RxJava 间隔
【发布时间】：2016-01-24 04:56:32
【问题描述】：

我对 rxjava 还很陌生，所以我想将它与 android IntentService 一起使用，并且我需要在特定时间段内每秒收到通知（类似于 Android CountDownTimer。我决定尝试使用 rxjava 并且我有这个类：

public class WorkoutService extends IntentService {
public static final String BUNDLE_EXTRA_MESSENGER = "messenger";
public static final String BUNDLE_EXTRA_NUMBER_ROUNDS = "nr_rounds";
public static final String BUNDLE_EXTRA_WORKOUT_DURATION = "workout_duration";
public static final String BUNDLE_EXTRA_PAUSE_DURATION = "pause_duration";
private static final int NOTIFICATION_ID = 1;
public static final int UPDATE_PROGRESS = 2;

/**
 * Target we publish for clients to send messages to IncomingHandler.
 * This is the messenger from the activity
 */
Messenger messenger;

private NotificationManager notifyManager;
private NotificationCompat.Builder builder;
private volatile int maxProgress;
private int numberOfRounds = 4;
private int workoutDuration = 7 * 60; //7 minutes
private int pauseDuration = 90; //1.5 minutes
private int currentProgress;

public WorkoutService() {
    super("WorkoutService");
}

@Override
protected void onHandleIntent(Intent intent) {
    Bundle extras = intent.getExtras();
    if (extras != null) {
        messenger = (Messenger) extras.get(BUNDLE_EXTRA_MESSENGER);
        numberOfRounds = extras.getInt(BUNDLE_EXTRA_NUMBER_ROUNDS, numberOfRounds);
        workoutDuration = extras.getInt(BUNDLE_EXTRA_WORKOUT_DURATION, workoutDuration);
        pauseDuration = extras.getInt(BUNDLE_EXTRA_PAUSE_DURATION, pauseDuration);
    }
    maxProgress = numberOfRounds * workoutDuration + ((numberOfRounds - 1) * pauseDuration);
    maxProgress = 10; //TODO: for testing
    showNotification(maxProgress);
    Timber.d("maxProgress %d", maxProgress);
    startWorkout();

}

private void startWorkout() {
    final Observable<Long> observable = Observable
            .interval(1, TimeUnit.SECONDS);
    observable
            .subscribeOn(Schedulers.io())
            .subscribe(new Subscriber<Long>() {
                @Override
                public void onCompleted() {
                    Timber.d("onCompleted");
                    unsubscribe();
                    stopForeground(true);
                    stopSelf();

                }

                @Override
                public void onError(Throwable e) {
                    Timber.e("onError");
                }

                @Override
                public void onNext(Long aLong) {
                    Timber.d("onNext : " + aLong + "S");
                    updateProgress();
                    if (aLong == maxProgress) {
                        onCompleted();
                    }
                }
            });
}

private void showNotification(int maxProgress) {
    Intent notificationIntent = new Intent(this, WorkoutService.class);
    PendingIntent pendingIntent = PendingIntent.getActivity(this, 0, notificationIntent, 0);
    notifyManager =
            (NotificationManager) getSystemService(Context.NOTIFICATION_SERVICE);
    builder = new NotificationCompat.Builder(this);
    builder.setContentTitle(getString(R
            .string.notification_title))
            .setContentIntent(pendingIntent)
            .setSmallIcon(R.mipmap.ic_launcher);
    startForeground(NOTIFICATION_ID, builder.build());
    currentProgress = 0;
    builder.setProgress(maxProgress, currentProgress, false);
    notifyManager.notify(NOTIFICATION_ID, builder.build());
}

private void sendMessageToActivity(Message message) {
    try {
        if (messenger != null) {
            messenger.send(message);
        }
    } catch (RemoteException e) {
        Timber.e(e, "Error sending message to activity");
    }
}

private void updateProgress() {
    currentProgress++;
    builder.setProgress(maxProgress, currentProgress, false);
    notifyManager.notify(NOTIFICATION_ID, builder.build());
    Message message = Message.obtain(null, UPDATE_PROGRESS, currentProgress, 0);
    sendMessageToActivity(message);
}

}

问题是通知没有被解除，而它应该被解除，即使我明确调用 stopSelf()，服务似乎也没有停止。在 android 文档中，它说当不再有工作要做时，该服务会自行停止，但是由于我正在调用 onCompleted 并取消订阅，难道不是这种情况吗？如何确保 observable 停止发射并终止流？非常感谢

【问题讨论】：

我可以看到很多问题。您应该使用takeWhile 来应用您的布尔条件并且不自己调用onCompleted，完成后您也已取消订阅，因此也无需调用它.此外，我认为您不应该手动调用 stopSelf 与意图服务一起使用，但我可能是错的

标签： android service rx-java

【解决方案1】：

问题是，当 onHandleIntent 返回时，您的意图服务已经死亡。 IntentServices 是一种非常特殊的服务，它在后台线程中执行 onHandleIntent 并被解除。

通过这样做，您会泄漏该意图服务类，因为订阅者持有对它的引用。订阅完成后，您将在死（泄漏）服务上调用 stopSelf。

此外，由于 onHandleIntent 本身在不同的线程中运行，因此在不同的线程中订阅是没有意义的。

我认为您应该使用服务（而不是意图服务）来实现您想要做的事情。

【讨论】：

我该怎么做

【解决方案2】：

感谢 fedepaol 和 David Medenjak，我设法解决了这个问题。作者：

切换到服务而不是意图服务
将 takeWhile 用于布尔条件

这是生成的类

public class WorkoutService extends Service {
public static final String BUNDLE_EXTRA_MESSENGER = "messenger";
public static final String BUNDLE_EXTRA_NUMBER_ROUNDS = "nr_rounds";
public static final String BUNDLE_EXTRA_WORKOUT_DURATION = "workout_duration";
public static final String BUNDLE_EXTRA_PAUSE_DURATION = "pause_duration";
private static final int NOTIFICATION_ID = 1;
public static final int UPDATE_PROGRESS = 2;

/**
 * Target we publish for clients to send messages to IncomingHandler.
 * This is the messenger from the activity
 */
Messenger messenger;

private NotificationManager notifyManager;
private NotificationCompat.Builder builder;
private volatile int maxProgress;
private int numberOfRounds = 4;
private int workoutDuration = 7 * 60; //7 minutes
private int pauseDuration = 90; //1.5 minutes
private int currentProgress;

@Override
public int onStartCommand(Intent intent, int flags, int startId) {
    Bundle extras = intent.getExtras();
    if (extras != null) {
        messenger = (Messenger) extras.get(BUNDLE_EXTRA_MESSENGER);
        numberOfRounds = extras.getInt(BUNDLE_EXTRA_NUMBER_ROUNDS, numberOfRounds);
        workoutDuration = extras.getInt(BUNDLE_EXTRA_WORKOUT_DURATION, workoutDuration);
        pauseDuration = extras.getInt(BUNDLE_EXTRA_PAUSE_DURATION, pauseDuration);
    }
    maxProgress = numberOfRounds * workoutDuration + ((numberOfRounds - 1) * pauseDuration);
    maxProgress = 10; //TODO: for testing
    showNotification(maxProgress);
    Timber.d("maxProgress %d", maxProgress);
    startWorkout();
    return START_STICKY;
}

@Nullable
@Override
public IBinder onBind(Intent intent) {
    return null;
}


private void startWorkout() {
    final Observable<Long> observable = Observable
            .interval(1, TimeUnit.SECONDS)
            .takeWhile(new Func1<Long, Boolean>() {
                @Override
                public Boolean call(Long aLong) {
                    return aLong <= maxProgress;
                }
            });
    observable.observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(new Subscriber<Long>() {
                @Override
                public void onCompleted() {
                    Timber.d("onCompleted");
                    stopForeground(true);
                    stopSelf();

                }

                @Override
                public void onError(Throwable e) {
                    Timber.e("onError");
                }

                @Override
                public void onNext(Long aLong) {
                    Timber.d("onNext : " + aLong + "S");
                    updateProgress();
                }
            });
}

private void showNotification(int maxProgress) {
    Intent notificationIntent = new Intent(this, WorkoutService.class);
    PendingIntent pendingIntent = PendingIntent.getActivity(this, 0, notificationIntent, 0);
    notifyManager =
            (NotificationManager) getSystemService(Context.NOTIFICATION_SERVICE);
    builder = new NotificationCompat.Builder(this);
    builder.setContentTitle(getString(R
            .string.notification_title))
            .setContentIntent(pendingIntent)
            .setSmallIcon(R.mipmap.ic_launcher);
    startForeground(NOTIFICATION_ID, builder.build());
    currentProgress = 0;
    builder.setProgress(maxProgress, currentProgress, false);
    notifyManager.notify(NOTIFICATION_ID, builder.build());
}

private void sendMessageToActivity(Message message) {
    try {
        if (messenger != null) {
            messenger.send(message);
        }
    } catch (RemoteException e) {
        Timber.e(e, "Error sending message to activity");
    }
}

private void updateProgress() {
    currentProgress++;
    builder.setProgress(maxProgress, currentProgress, false);
    notifyManager.notify(NOTIFICATION_ID, builder.build());
    Message message = Message.obtain(null, UPDATE_PROGRESS, currentProgress, 0);
    sendMessageToActivity(message);
}

}

【讨论】：