Android.os.NetworkOnMainThreadException с помощью rxjava на Android

Ответ 1

Observable.just(...) вызывается сразу в вызывающем потоке (в этом случае основной поток). Ваш код - это просто встроенная версия:

boolean activeConn = Utils.isActiveInternetConnection(Launcher.this);
AndroidObservable.bindActivity(this, 
        Observable.just(activeConn))
        .subscribeOn(...)
        ...

Вы попытались переместить его из основного потока, вызвав subscribeOn() - но вызов уже произошел.

Как мы справляемся с этим (и я не уверен, что это лучший способ, но он работает) состоит в том, чтобы отложить сеть или заблокировать вызов до тех пор, пока не произойдет подписка, настройте наблюдаемое для выполнения на правильных потоках и затем подпишитесь:

AndroidObservable.bindActivity(this,
        Observable.defer(new Func0<Boolean>() {
            @Override
            public Observable<Observable<Boolean>> call() {
                return Observable.just(Utils.isActiveInternetConnection(Launcher.this));
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<Boolean>() {
            @Override
            public void call(Boolean aBoolean) {
                if (aBoolean) {
                    Toast.makeText(Launcher.this, "There is internet connection", Toast.LENGTH_SHORT).show();
                } else {
                    Toast.makeText(Launcher.this, "There is no internet connection", Toast.LENGTH_SHORT).show();
                }
            }
        });

Ответ 2

Я думаю, just просто вызывает метод синхронно, поскольку он ожидает логическое значение и пытается его получить.

Я плохо себя чувствую при RxJava, но вы можете попробовать что-то вроде этого:

Observable<Boolean> onlineObservable = Observable.create(new Observable.OnSubscribe<Boolean>() {
    @Override
    public void call(Subscriber subscriber) {
        subscriber.onNext(Utils.isActiveInternetConnection(context));
    }
});
onlineObservable.subscribeOn(Schedulers.newThread()).subscribe(result -> {...});

Ответ 3

это мои данные из базы данных с помощью кода RXAndroid:

    Observable.create(new Observable.OnSubscribe<List<GRO_VO>>() {
        @Override
        public void call(Subscriber<? super List<GRO_VO>> subscriber) {

            String jsonIn;
            jsonIn =retrieveDataFromDB();
            Gson gson = new Gson();
            Type listType = new TypeToken<List<GRO_VO>>() {

            }.getType();   
            eventJoinList = gson.fromJson(jsonIn, listType);
            Log.d("RX",jsonIn);
            subscriber.onNext(eventJoinList);
        }
    })

                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<List<GRO_VO>>() {
        @Override
        public void call(List<GRO_VO> eventJoinList) {
            Log.d("RX", ".subscribe");
       recyclerView.setAdapter(new EventJoinAdapter(eventJoinList));
        }
    });

Я думаю, что оператор just немедленно выдаст данные, поэтому нецелесообразно извлекать данные из базы данных через сеть. Он очень прост в использовании, но его можно использовать только для данных, уже находящихся в мобильном телефоне.

У меня также была такая проблема, как у @Baniares, но после того, как я использую оператор create, проблема исчезает...

Из документации RXJava:

static <T> Observable<T> create(Observable.OnSubscribe<T> f)

Возвращает наблюдаемое, которое будет выполнять указанную функцию, когда подписчик подписывается на нее.

Используя оператор create, можно установить стандартный процесс:

1 .subscribe(...) Подписчик (подкласс класса Observer) запустит соединение с Observable.

2 .subscribeOn(Schedulers.io()) собрать обратный поток из RX-ThreadPool

3 .create(...) извлекать данные с сервера... во время некоторого netWork..etc

4 .observeOn(AndroidSchedulers.mainThread()) это означает, что данные будут заданы с помощью UI-Thread

5 Как только мы получим данные, мы можем установить данные в методе onNext() в .subscribe( ), данные будут установлены в пользовательском интерфейсе UI-Thread, поскольку мы создаем UI-поток для работы над .observerOn(AndroidSchedulers.mainThread())

цепочка методов просто не выполняется порядком кода.

Обратите внимание, что если вы используете оператор .create(), вы должны закончить свое наблюдение в .create(), другой оператор, такой как map, flatMap не будет выполнен после оператора .create().

Ответ 4

AdamS правильно, однако RxJava 2 теперь предлагает Observable.fromCallable() отложить наблюдаемую операцию до подписки. Хорошая рекомендация: https://caster.io/lessons/fromcallable-converting-slow-methods-into-an-observable/

Пример кода примера из моего прецедента:

Single.fromCallable(new Callable<Response>() {
        @Override
        public Response call() throws Exception {
            return NetworkGateway.networkRequest();
        }
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(observer);