英文:
RxJava: how to make code working asynchronously?
问题
请纠正我对异步工作的结论是否正确,并提出任何关于如何使这个工作异步的建议。
这段代码执行了将计算结果设置到网格单元格中。根据我的理解,异步工作应该分批次显示网格单元格。所以如果我们有8个核心,那么结果就可以看到8个单元格,然后在一段时间后再看到另外8个,依此类推(如果我设置了一些时间延迟的话)。但目前的情况是,结果单元格一个接一个地显示出来。
Model(模型):
public class ListRepository implements ListRepositoryInterface {
// ...(其他代码部分不翻译)
}
Presenter(呈现者):
public void calculate(int inputNumber) {
// ...(其他代码部分不翻译)
}
UPD(更新):
private void run() {
// ...(其他代码部分不翻译)
}
// ...(其他代码部分不翻译)
Gradle:
dependencies {
implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
implementation 'io.reactivex.rxjava3:rxjava:3.0.4'
}
如果您需要进一步的翻译或解释,请提出具体的问题。
英文:
Please, coorect me if I wrong in my conclusions about async work and give any suggestions how to make this work asynchronously.
This code performs setting result of calculations in grid cells.
As I assume asynchronous work should display grid cells by portions. So if we have 8 cores as result we can see that 8 cells displayed and during some time another 8 and so on (in case if I set some time delay). But for now as result cells displaying become one by one.
Model:
public class ListRepository implements ListRepositoryInterface {
private final Integer insertValue = 1000000;
private HashMap<String, BaseUnit> unitMap = new HashMap<>();
@Inject
public ListRepository() {}
public PublishSubject<BaseUnit> exec(int inputNumber) {
PublishSubject<BaseUnit> subject = PublishSubject.create();
Observable<BaseUnit> observable = getListObservable(inputNumber)
.subscribeOn(Schedulers.computation())
.flatMap(resultList ->
Observable.fromIterable(resultList)
.flatMap(listElem ->
Observable.fromArray(ListOperationName.values())
.map(operationElem -> {
ListUnit unit = new ListUnit(operationElem, listElem, 0);
calculate(unit, listElem);
unitMap.put(unit.getViewId(), unit);
return unit;
})
)
);
observable.subscribe(subject);
return subject;
}
private Observable<ArrayList<List<Integer>>> getListObservable(int inputNumber) {
return Observable.fromCallable(() -> {
ArrayList<List<Integer>> list = new ArrayList<>();
Integer[] populatedArray = new Integer[inputNumber];
Arrays.fill(populatedArray, insertValue);
list.add(new ArrayList<>(Arrays.asList(populatedArray)));
list.add(new LinkedList<>(Arrays.asList(populatedArray)));
list.add(new CopyOnWriteArrayList<>(Arrays.asList(populatedArray)));
return list;
});
}
private void calculate(ListUnit unit, List<Integer> list) {
try {
TimeUnit.MILLISECONDS.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
double start = getTime();
//noinspection SynchronizationOnLocalVariableOrMethodParameter
synchronized (list) {
switch (unit.getOperationName()) {
case ADD_FIRST:
list.add(0, insertValue);
break;
case ADD_MID:
list.add(list.size() / 2, insertValue);
break;
case ADD_LAST:
list.add(insertValue);
break;
case SEARCH:
//noinspection unused
boolean contains = list.contains(insertValue);
break;
case RM_FIRST:
list.remove(0);
break;
case RM_MID:
list.remove(list.size() / 2);
break;
case RM_LAST:
list.remove(list.size() - 1);
break;
}
}
unit.setTime(getTime() - start);
}
private double getTime() {
return System.nanoTime();
}
public HashMap<String, BaseUnit> getUnitMap() {
return unitMap;
}
}
Presenter:
public void calculate(int inputNumber) {
fragment.showAllProgressBars();
PublishSubject<BaseUnit> subject = repository.exec(inputNumber);
Disposable disposable = subject.observeOn(AndroidSchedulers.mainThread())
.subscribe(unit -> {
fragment.setCellText(unit.getViewId(), unit.getTimeString());
}, Throwable::printStackTrace);
}
UPD: now I made test example and trying to subscribe to subject correctly.
If I use subject.onNext() it works asynchronously, but I assume this is wrong because it doesn't get to check subject.hasComplete() and get "true".
See comments "TODO" below.
private void run() {
Log.d("APP", "INIT");
PublishSubject<String> subject = exec(1000000);
subject.observeOn(AndroidSchedulers.mainThread())
.subscribe(unit -> {
Log.d("STRING RESULT = ", unit);
if (subject.hasComplete()) {
//TODO: this condition should work if we use observable.subscribe(subject), not subject.onNext
Log.d("SUBJECT", "COMPLETED");
}
}, Throwable::printStackTrace);
}
private int insertValue = 1000000;
public PublishSubject<String> exec(int inputNumber) {
PublishSubject<String> subject = PublishSubject.create();
getListObservable(inputNumber)
.flatMap(resultList -> getOperationsObservable()
.flatMap(operationElem -> getResultListObservable(resultList)
.map(listElem ->
calculate(operationElem, listElem)
//TODO: should be smth like this, i.e. calculate.subscribe(subject)
// .subscribe(subject)
.subscribe(subject::onNext)
)
)
).subscribe();
return subject;
}
private Observable<ArrayList<List<Integer>>> getListObservable(int inputNumber) {
return Observable.fromCallable(() -> {
ArrayList<List<Integer>> list = new ArrayList<>();
Integer[] populatedArray = new Integer[inputNumber];
Arrays.fill(populatedArray, insertValue);
list.add(new ArrayList<>(Arrays.asList(populatedArray)));
list.add(new LinkedList<>(Arrays.asList(populatedArray)));
list.add(new CopyOnWriteArrayList<>(Arrays.asList(populatedArray)));
return list;
}).subscribeOn(Schedulers.computation());
}
private Observable<String> calculate(ListOperationName operationName, List<Integer> list) {
return Observable.fromCallable(() -> {
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
double start = getTime();
synchronized (list) {
switch (operationName) {
case ADD_FIRST:
list.add(0, insertValue);
break;
case ADD_MID:
list.add(list.size() / 2, insertValue);
break;
case ADD_LAST:
list.add(insertValue);
break;
case SEARCH:
//noinspection unused
boolean contains = list.contains(insertValue);
break;
case RM_FIRST:
list.remove(0);
break;
case RM_MID:
list.remove(list.size() / 2);
break;
case RM_LAST:
list.remove(list.size() - 1);
break;
}
}
return Double.toString(getTime() - start);
}).subscribeOn(Schedulers.newThread());
}
private Observable<ListOperationName> getOperationsObservable() {
return Observable.fromArray(ListOperationName.values());
}
private Observable<List<Integer>> getResultListObservable(ArrayList<List<Integer>> list) {
return Observable.fromIterable(list);
}
private double getTime() {
return System.nanoTime();
}
public enum ListOperationName {
ADD_FIRST,
ADD_MID,
ADD_LAST,
SEARCH,
RM_FIRST,
RM_MID,
RM_LAST;
}
Gradle:
dependencies {
implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
implementation 'io.reactivex.rxjava3:rxjava:3.0.4'
}
答案1
得分: 1
如果我从calculate(...)
方法中移除.subscribeOn(Schedulers.newThread());
,并对exec(...)
方法进行以下更改,我似乎可以获得所期望的并行性。
public Observable<String> exec(int inputNumber) {
return getListObservable(inputNumber)
.flatMap(resultList -> getOperationsObservable()
.flatMap(operationElem -> getResultListObservable(resultList)
.flatMap(listElem -> Observable.just(listElem)
.subscribeOn(Schedulers.computation())
.flatMap(__ -> calculate(operationElem, listElem)))));
}
我移除了PublishSubject
- 这似乎是多余的,你可以直接订阅生成的Observable
,不是吗?
英文:
If I remove .subscribeOn(Schedulers.newThread());
from the calculate(...)
method in your most recent example and make the following change to exec(...)
, I seem to get the desired parallelism.
public Observable<String> exec( int inputNumber )
{
return getListObservable( inputNumber )
.flatMap( resultList -> getOperationsObservable()
.flatMap( operationElem -> getResultListObservable( resultList )
.flatMap( listElem -> Observable.just( listElem )
.subscribeOn( Schedulers.computation() )
.flatMap( __ -> calculate( operationElem, listElem )))));
}
I removed PublishSubject
- it seems redundant, you can just subscribe to the resulting Observable
, no?
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论