我是 rxjava
或 rxandroid 的新手,我想将我的 AsyncTask
和回调 hell 更改为 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 onError(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 onError(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();
}
});
}
关于android - flatMap 中的 NetworkOnMainThreadException,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/41633226/