如何使用 java8 和 rxjava 组合 Hystrix observables

Posted 2023-03-16

【中文标题】如何使用 java8 和 rxjava 组合 Hystrix observables

【英文标题】：How to compose Hystrix observables using java8 and rxjava

【发布时间】：2015-10-01 20:59:26

【问题描述】：

我有两个要调用的 api。它们都用 Hystrix Observable 包裹：

这里有一个例子：

@HystrixCommand()
    public Observable<String> getAvailableFlightBookings() 
        return new ObservableResult<String>() 
            @Override
            public String invoke() 
                URI uri = registryService.getServiceUrl("flight-booking-service", "http://localhost:8081/flight-booking-service");
                String url = uri.toString() + "/flights/list";
                ResponseEntity<String> resultStr = restTemplate.getForEntity(url, String.class);
                return resultStr.getBody();
            
        ;
    

我正在努力完成以下代码：

public DeferredResult<FlightDetails> getAllFlightDetails() 
        //Calling previous defined functions
        Observable<String> availableFlightBookings=flightBookingIntegrationService.getAvailableFlightBookings();
        Observable<String> couponId=couponIntegrationService.getCoupon();


        Observable<String> details = Observable.zip(

    ..?


    

我不确定如何调用以下 API：

flightBookingIntegrationService.getAvailableFlightBookings();
couponIntegrationService.getCoupon();

并使用 Observable.zip 填充最终结果对象 (FlightDetails)

航班详情：

public class FlightDetails 

    String couponId;
    String availableFlightList;

..

谢谢你， 射线。

【参考方案1】：

首先将您的 Hystrix 命令类设为 @Component，因为您使用的是 Spring，然后只需将 Autowire 放入您的控制器并调用

使用 lambdas，看起来像这样：

public DeferredResult<FlightDetails> getAllFlightDetails() 

    Observable<String> availableFlightBookings=flightBookingIntegrationService.getAvailableFlightBookings();
    Observable<String> couponId=couponIntegrationService.getCoupon();

    //Create a new DeferredResult
    DeferredResult<FlightDetails> result = new DeferredResult();

    Observable.zip(availableFlightBookings,couponId, (avaliable, coupon) -> 
     // do some logic here or just..
    return new FlightDetails(avaliable,coupon);
    ).subscribe(result::setResult,result::setErrorResult);
    return result;

【讨论】：

顺便说一句，如果您只想创建 FlightDetails 的新实例，您可以使用方法引用而不是 lambda

你能解释一下为什么使用 DeferredResult 吗？

以及使用(avaliable, coupon)的意思。

首先，因为你已经在使用，而且，这会触发spring的异步处理，所以你不需要阻塞Observable得到结果，Spring Cloud会带来返回的可能可从控制器观察，然后，您不需要延迟，这将由框架完成

(available, coupon) 只是您需要传递 zip 操作的函数的参数名称，因为我们使用的是 Java 8 lambda，所以我们不需要匿名调用，你可以随意调用变量

【参考方案2】：

我对 Hystrix 不熟悉，但是压缩两个 observable 应该与纯 RxJava 没有什么不同。

Observable.zip(availableFlightBookings, couponId, new Func2<String, String, FlightDetails>() 
    @Override
    public FlightDetails call(String availableFlights, String coupon) 
        return new FlightDetails(availableFlights, coupon);
    
).subscribe();