RxJava 가입

Question

나는이 질문, 간단 비즈니스 로직 흐름입니다 : 여러 부서, 부서 및 직원 관계 직원이 캐시에 있는지 여부를

검사, 첫번째 경우, 관계가 캐시에 존재하는지 확인 존재한다면, 캐시에없는 경우 직원이 속한 지 여부를 확인하고 데이터베이스에서 가져 와서 직원에 대한 관계를 확인한 다음 부서 정보를 캐시에 저장하십시오.

public Observable isEmployeeInDepartment(List<Long> departmentIds, long employeeId){ 

    //this observable will resolve twice, and cause unnecessary cache access 
    Observable departmentInfoExsitInCache= checkDepartmentInfoFromCache(...).share(); 

    Observable departInfoNotInCache = departmentInfoExsitInCache.filter(...); 

    //this observable will resolve twice, and cause unnecessary database access 
    Observable departmentInfoFromDb=departInfoNotInCache.flatMap(departmentIds->checkFromDb()).share(); 

    Observable<Long> saveResult=departmentInfoFromDb.flatMap(departmentInfo->saveToCache()); 

    Observable<Long> departInfoInCache = departmentInfoExsitInCache.filter(...); 

    return departInfoInCache.check(userId).merge(departmentInfoFromDb.check(userId)).doOnCompleted(saveResult.subscribe()); 
} 

문제가 departmentInfoExsitInCache 및 saveResult 클라이언트 방법은 구독 한 번으로 두 번 해결 될 것입니다 :

이

는 코드입니다.

한번 저장 구독 코드 .doOnCompleted (saveResult.subscribe())를 제거하면 정상이되어 한 번만 해결된다는 것을 알았습니다. 이 코드에 문제가 있습니까?

Answer 1

여기 공유를 잘못 사용하는 것입니다.
이 경우 share()이 도움이되지 않습니다. share()은 실제로 publish().autoConnect()이 스트림에 대한 단일 구독을 유지하기 때문에 구독을 다시 호출해도 구독 논리가 다시 호출되지는 않지만 기존 스트림에 연결됩니다.
공유 스트림에서 구독을 취소 한 모든 사용자가 Observable을 구독 취소하면 다시 subscribe()으로 전화하면 구독 논리를 호출하고 DB/캐시를 다시 호출합니다.

그래서 구독 취소 후 공유 운영자에게 다시 구독하는 것입니다. (doOnCompleted()에서) departmentInfoFromDb 및 departmentInfoExsitInCache이 다시 구독하고 DB/Cache로 이동하게됩니다.

cache()/reply() 연산자를 사용하여 subscription간에 DB/Cache에서 가져온 값을 유지하는 것이 좋습니다.