Observable wird weiterhin auf dem UI-Thread ausgeführt.

Question

Ich habe ein Problem mit der Ausführung meiner RxJava-Observablen in einem neuen Thread. Ich versuche gezippte DB-Abfragen unter Scheduler.io() auszuführen, blockiert jedoch den UI-Thread unabhängig davon. Nicht sicher warum.

-Code für DB:

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms, 
     KeyMatchLevel keyMatchLevel, int queryLim) { 
    return Observable.just(
      searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim)); 
} 

Hier ist der Code, wo das Zuziehen passiert:

public Observable<Set<Airport>> getSearchResultsForAll(String term, int queryLim) { 
    List<String> terms = CollectionsUtil.asArrayList(term); 
    Observable<List<Airport>> macObservable = onNewThread(
      searchModelsInView(getMacView(), terms, 
        KeyMatchLevel.PREFIXED, queryLim)); 
    Observable<List<Airport>> nameObservable = onNewThread(
      searchModelsInView(getNameView(), terms, 
        KeyMatchLevel.PREFIXED, queryLim)); 
    Observable<List<Airport>> codeObservable = onNewThread(
      searchModelsInView(getCodeView(), terms, 
        KeyMatchLevel.PREFIXED, NONE)); 
    Observable<List<Airport>> regionObservable = onNewThread(
      searchModelsInView(getCityView(), terms, 
        KeyMatchLevel.PREFIXED, NONE)); 

    ; 
    return Observable.zip(macObservable, nameObservable, codeObservable, regionObservable, 
      (macList, nameList, codeList, regionList) -> { 

       Set<Airport> resultSet = new LinkedHashSet<Airport>(); 

       Airport mac = macList.get(0); 

       if (term.length() > CODE_LEN) { 
        if (lowerCase(mac.getCity()).contains(lowerCase(term))) { 
         handleMACFound(resultSet, mac, regionList); 
         return resultSet; 
        } 
        resultSet.addAll(nameList); 
        resultSet.addAll(regionList); 
        return resultSet; 
       } 


       boolean macFound = StringUtil.isEquals(lowerCase(term), 
         lowerCase(mac.getCode())); 

       if (macFound) { 
        handleMACFound(resultSet, mac, codeList); 
        return resultSet; 
       } 

       resultSet.addAll(codeList); 
       resultSet.addAll(nameList); 
       resultSet.addAll(regionList); 

       return resultSet; 
      }); 
} 

private Observable<List<Airport>> onNewThread(Observable<List<Airport>> observable) { 
    return observable.subscribeOn(Schedulers.io()); 
} 

Dies ist, wo es heißt:

public void searchKey(String searchTerms, Airport... excluding) { 
    TBDataBase.getAirportDB() 
      .getSearchResultsForAll(searchTerms, MAX_AIRPORT_SEARCHED) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(airports -> { 
       setResults(airports, excluding); 
       listener.onRegularAirportSearch(); 
       listener.onSearchPerformed(); 
      }); 


} 

Es scheint, wie es auf der laufen soll Scheduler.io() aber es blockiert den Haupt-Thread, so dass es nicht funktioniert.

Answer 1

Es ist wie das Problem aussieht, ist in Ihrem searchModelsTask:

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms, 
     KeyMatchLevel keyMatchLevel, int queryLim) { 
    return Observable.just(
      searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim)); 
} 

Observable.just erwartet, dass Sie bereits den Wert haben. Daher wird searchModelsTask ausgeführt, bevor das Observable überhaupt erstellt wird. Daher spielt es keine Rolle, dass Sie versuchen, es auf den I/O-Scheduler zu setzen. Versuchen Sie es mit einer Zurückstellungs mit dem gerade innen:

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms, 
     KeyMatchLevel keyMatchLevel, int queryLim) { 
    return Observable.defer(() -> Observable.just(
      searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim))); 
} 

oder

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms, 
      KeyMatchLevel keyMatchLevel, int queryLim) { 

return Observable.defer(new Func0<Observable<List<T>>>(){ 

     @Override 
     public Observable<List<T>> call() { 
      return Observable.just(
        searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim)); 
     } 
    }); 
} 

Sie auch als auch in mit Observable.callable statt defer aussehen könnte.