android - 尽管实现了doOnError，但RxAndroid会崩溃应用程序

我有以下代码。这基本上是试图将特定sqlite表中的所有数据发送到dynamodb：

Observable.create(new Observable.OnSubscribe<Area>() {
    @Override
    public void call(Subscriber<? super Area> subscriber) {
        try {
            for (Area item : areaDao.listAll()) {
                subscriber.onNext(item);
            }
            subscriber.onCompleted();
        } catch (Exception e) {
            subscriber.onError(e);
        }
    }
}).flatMap(new Func1<Area, Observable<Area>>() {
    @Override
    public Observable<Area> call(Area area) {
        dynamoDBMapper.save(area);
        return Observable.just(area);
    }
}).observeOn(
        AndroidSchedulers.mainThread()
).doOnError(new Action1<Throwable>() {
    @Override
    public void call(Throwable throwable) {
        Log.w("AreaHandler", "Could not upload area", throwable);
    }
}).doOnCompleted(new Action0() {
    @Override
    public void call() {
        Toast.makeText(ctx, R.string.toast_upload_successful, Toast.LENGTH_SHORT).show();
    }
}).subscribeOn(
        Schedulers.io()
).subscribe(new Action1<Area>() {
    @Override
    public void call(Area area) {
        areaDao.delete(area.getId());
    }
});

我试图在一个禁用了互联网连接的模拟器上运行它，结果是dynamo客户端进行了几次（失败的）重试，然后抛出一个异常并使应用程序崩溃。从我在文档中读到的内容来看，这个异常应该被doonerror所吞没，而不是被释放到野外，扼杀这个过程。
我错过了什么？

最佳答案

你在错误的地方抓住了错误。doOnError用于副作用。它不处理错误。
选项1。传入两个操作1

Observable.just(1, 2, 3)
            .subscribe(
                    new Action1<Integer>() {
                        @Override
                        public void call(Integer integer) {
                            System.out.println(integer);
                        }
                    },
                    new Action1<Throwable>() {
                        @Override
                        public void call(Throwable throwable) {
                            System.err.print(throwable);
                        }
                    });

选项2：传入一个观察者

    Observable.just(1, 2, 3)
            .subscribe(new Observer<Integer>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable throwable) {
                    System.err.print(throwable);
                }

                @Override
                public void onNext(Integer integer) {
                    System.out.println(integer);
                }
            });