rxjava - 获取响应并且并行插入差异

我使用rxjava来使用Observable.zip并行处理两个请求。我试图做的是，在一个observable say response我得到一个响应，并在其他observable say diff我想获得响应，并保存在数据库中的差异。问题是我不知道如何实现我的要求作为diff observable是没有得到完成，如果response observable获得响应rxjava - 获取响应并且并行插入差异

下面是我在做什么...

public ServiceResponse getDummyResponse(ServiceRequest serviceRequest, String prodId){ 
    Observable<ServiceResponse> subInfoDummyObservable = getDummyResonseGenericObservable(); 
    Observable<ServicesDiff> reObservable = getServicesDiffGenericObservable(serviceRequest, prodId); 

    Observable<ServiceResponse> responseObservable = Observable.zip(
      subInfoDummyObservable, 
      reObservable, 
      new Func2<ServiceResponse, ServicesDiff, ServiceResponse>() { 
       @Override 
       public ServiceResponse call(ServiceResponse serviceResponse, ServicesDiff diffResponse) { 
        return serviceResponse; 
       } 
      } 
    ); 

    ServiceResponse serviceResponse = responseObservable.toBlocking().single(); 

    return serviceResponse; 
} 

Observable<ServiceResponse> getDummyResonseGenericObservable() { 
    return GenericHystrixCommand.toObservable("getDummyResonseGenericObservable", "getDummyResonseGenericObservable",() -> new ServiceResponse(),(t) -> {return null;}); 
} 

Observable<ServicesDiff> getServicesDiffGenericObservable(ServiceRequest serviceRequest, String prodId) { 
    return GenericHystrixCommand.toObservable("getServicesDiffGenericObservable", "getServicesDiffGenericObservable",() -> getBothServiceResponses(serviceRequest, prodId),(t) -> {return null;}); 
} 

public ServicesDiff getBothServiceResponses(ServiceRequest serviceRequest, String prodId) { 
    Observable<String> service1ResponseObservable = getService1GenericObservable(prodId); 
    Observable<ServiceResponse> service2ResponseObservable = getService2GenericObservable(serviceRequest, prodId); 

    Observable<ServicesDiff> observable = Observable.zip(
      service1ResponseObservable, service2ResponseObservable, 
      new Func2<String, ServiceResponse, ServicesDiff>() { 
       @Override 
       public ServicesDiff call(String service1Response, ServiceResponse service2Response) { 
        return aggregate(service1Response, service2Response); // never reaches this point********** 
       } 
      } 
    ); 
    ServicesDiff response = observable.toBlocking().single(); 

    return response; 
}

我插入diff来DB在aggregate方法，但它永远不会达到aggregate。请让我知道我在这里做错了什么？谢谢。

来源

2017-02-10 user1653027

你的代码示例是我不明白，是什么问题所在的getBothServiceResponses方法与代码的其余部分之间的关系？你在getBothServiceResponses上压缩的两个观察对象是什么？ – yosriz

我同意，代码需要澄清。前三种方法永远不会被调用，所以我们不知道第四个方法'getBothServiceResponses（）'中的观察值是什么样子的，问题出在哪里 – nosyjoe

Observable是如何使用数据的说明。在您的代码示例中，您不需要订阅，也不会实际使用数据。您刚刚描述了如何请求，但是订阅部分（即触发请求的部分）缺失。

所以，如果我重写了一点代码：

class Aggregate { 
    Aggregate(String reponse, ServicesDiff diff) { 
     ... 
    } 
} 

Observable<String> getService1GenericObservable(String prodId) { 
    ... 
} 

Observable<ServicesDiff> getServicesDiffGenericObservable(ServiceRequest serviceRequest, String prodId) { 
    ... 
} 

public Observable<Aggregate> getBothServiceResponses(ServiceRequest serviceRequest, String prodId) { 
    Observable<String> service1ResponseObservable = getService1GenericObservable(prodId); 
    Observable<ServiceResponse> service2ResponseObservable = getService2GenericObservable(serviceRequest, prodId); 

    return Observable<Aggregate> observable = Observable.zip(
      service1ResponseObservable, service2ResponseObservable, 
      new Func2<String, ServiceResponse, ServicesDiff>() { 
       @Override 
       public ServicesDiff call(String service1Response, ServiceResponse service2Response) { 
        return aggregate(service1Response, service2Response); 
       } 
      } 
    ); 
}

你将不得不这样做是为了访问结果汇总：

getBothServiceResponses(serviceRequest, prodId).subscribe(...)

来源

2017-02-13 16:26:02

rxjava - 获取响应并且并行插入差异

回答

相关问题