android-flatMap中的NetworkOnMainThreadException

我是rxjava或rxandroid的新手,我想将AsyncTask和回调地狱更改为Observable和Subscriber.但我对此有疑问.我需要对数据库执行两个请求.

我对第一个请求的答复将是第二个请求的结果.我尝试使用flatMap解决此问题.第一个请求返回值,一切正常,但是下一个请求给了我NetworkOnMainThreadException.

我知道该请求在主线程上执行,但是为什么呢？我尝试在flatMap之前添加subscribeOn(Schedulers.io()),但结果相同.您能帮我一下,并解释我做错了什么吗？提前致谢.我的代码…

private void getFavouriteList(){
    Observable.create((Observable.OnSubscribe<PaginatedScanList<UserDO>>) subscriber -> {
        final Map<String, AttributeValue> filterExpressionAttributeValues = new HashMap<>();
        filterExpressionAttributeValues
                .put(":val1", new AttributeValue().withS(sharedPreferences.getString("socialId", "")));
        final DynamoDBScanExpression scanExpression = new DynamoDBScanExpression()
                .withFilterExpression("socialId = :val1")
                .withExpressionAttributeValues(filterExpressionAttributeValues);
        PaginatedScanList<UserDO> result = dynamoDBMapper.scan(UserDO.class, scanExpression);
        Log.d(TAG, "first result size " + result.size());
        subscriber.onNext(result);
        subscriber.onCompleted();
    })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .filter(result -> {
                if(result.isEmpty()) {
                    Toast.makeText(context, "Can not find user", Toast.LENGTH_SHORT).show();
                    return false;
                }
                return true;
            })
            .flatMap(user -> Observable.from(user.get(0).getFavourites()))
            .subscribeOn(Schedulers.io())
            .flatMap(result -> {
                final Map<String, AttributeValue> filterExpressionAttributeValues = new HashMap<>();
                filterExpressionAttributeValues
                        .put(":val1", new AttributeValue().withS(result));
                filterExpressionAttributeValues
                        .put(":val2", new AttributeValue().withN("1"));
                final DynamoDBScanExpression scanExpression = new DynamoDBScanExpression()
                        .withFilterExpression("productId = :val1 and selling = :val2")
                        .withExpressionAttributeValues(filterExpressionAttributeValues);
                PaginatedScanList<ProductDO> res = dynamoDBMapper.scan(ProductDO.class, scanExpression);
                Log.d(TAG, "second result size " + res.size());
                return Observable.from(res);
            })
            .subscribe(new Subscriber<ProductDO>() {
                @Override
                public void onCompleted() {
                    favouriteProgressBar.setVisibility(View.INVISIBLE);
                }

                @Override
                public void one rror(Throwable e) {
                    e.printStackTrace();
                    favouriteProgressBar.setVisibility(View.INVISIBLE);
                }

                @Override
                public void onNext(ProductDO productDO) {
                    Log.d(TAG, "productId " + productDO.getProductId());
                    product.add(productDO);
                    adapter.notifyDataSetChanged();
                }
            });
}

解决方法:

将您的.observeOn(AndroidSchedulers.mainThread())移动到订阅之前.

PS：除非没有其他选择,否则不应使用Observable.create().

编辑以解决过滤器问题中的Toast.

private void getFavouriteList(){
    Observable.create((Observable.OnSubscribe<PaginatedScanList<UserDO>>) subscriber -> {
        final Map<String, AttributeValue> filterExpressionAttributeValues = new HashMap<>();
        filterExpressionAttributeValues
                .put(":val1", new AttributeValue().withS(sharedPreferences.getString("socialId", "")));
        final DynamoDBScanExpression scanExpression = new DynamoDBScanExpression()
                .withFilterExpression("socialId = :val1")
                .withExpressionAttributeValues(filterExpressionAttributeValues);
        PaginatedScanList<UserDO> result = dynamoDBMapper.scan(UserDO.class, scanExpression);
        Log.d(TAG, "first result size " + result.size());
        subscriber.onNext(result);
        subscriber.onCompleted();
    })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .filter(result -> {
                if(result.isEmpty()) {
                    Toast.makeText(context, "Can not find user", Toast.LENGTH_SHORT).show();
                    return false;
                }
                return true;
            })
            .observeOn(Schedulers.io())
            .flatMap(user -> Observable.from(user.get(0).getFavourites()))
            .flatMap(result -> {
                final Map<String, AttributeValue> filterExpressionAttributeValues = new HashMap<>();
                filterExpressionAttributeValues
                        .put(":val1", new AttributeValue().withS(result));
                filterExpressionAttributeValues
                        .put(":val2", new AttributeValue().withN("1"));
                final DynamoDBScanExpression scanExpression = new DynamoDBScanExpression()
                        .withFilterExpression("productId = :val1 and selling = :val2")
                        .withExpressionAttributeValues(filterExpressionAttributeValues);
                PaginatedScanList<ProductDO> res = dynamoDBMapper.scan(ProductDO.class, scanExpression);
                Log.d(TAG, "second result size " + res.size());
                return Observable.from(res);
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<ProductDO>() {
                @Override
                public void onCompleted() {
                    favouriteProgressBar.setVisibility(View.INVISIBLE);
                }

                @Override
                public void one rror(Throwable e) {
                    e.printStackTrace();
                    favouriteProgressBar.setVisibility(View.INVISIBLE);
                }

                @Override
                public void onNext(ProductDO productDO) {
                    Log.d(TAG, "productId " + productDO.getProductId());
                    product.add(productDO);
                    adapter.notifyDataSetChanged();
                }
            });
}

标签：rx-java,rx-android,android
来源： https://codeday.me/bug/20191111/2022732.html