使用rest调用的微服务中的Spring Reactive

Question

我使用spring boot web flux从事一个微服务项目，下面是一些服务：

• BaseInfoService
• 通知服务
• 会计服务
• 订单服务
• PerformService

@PostMapping("/create")
//@PostMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Mono<OrderDto> createOrder(@RequestParam OrderRequestDto orderRequest){
    return service.createOrder(orderRequest);
}

public Mono<OrderDto> createOrder(OrderRequestDto orderRequest){
    rule.validate(orderRequest);

    //====== in the same time =======//

    //@todo call baseInfoService to get some data

    //@todo call baseInfoService to get some other data

    //@todo call performService to send orderRequest some calculation

    //====== after getting above request ======//

    //@todo calculate and fill some order data acording above request results        

    OrderEntity order = transformer.transform(orderRequest);

    order = repository.save(order); 

    OrderDto orderDto = transformer.transformToDto(order);

    //======== after above calculation and save do these operation in the same time =======// 

    //@todo call accountService to effect the accounts
    //failure on this service is important and I should take proper action (there should be a call back instead of continue and sending notif)

    //@todo call notificationService to inform related users

    return Mono.just(orderDto);
}

现在，我应该使用webClient进行服务调用，将来我会使用kafka和spring cloud stream并将请求作为事件发送。

• 第一个问题是我应该在控制器中使用发布服务器 还是orderrequestdto？
• 第二个问题：我在github和其他站点中看到了许多简单的示例，但所有这些示例都只是从存储库中获取一个mono，然后将其传递给controller以返回给用户，而我找不到像这样复杂的现实世界示例。你能提供我应该如何实施它吗？

谢谢你。

Answer 1

首先，如果使用ractor，就不应该像在save方法中那样调用阻塞API。当您使用webflux时，您只有少量的线程，如果您阻止这些线程，您的应用程序性能将非常差。我建议改用反应型数据库驱动程序。

1,在控制器中不应该使用平面对象,因为必须阻塞线程才能得到对象本身。在反应器中，不能调用阻塞操作。如果你不确定什么是阻塞，也建议使用blockhound。如果调用阻塞方法，它将在测试期间抛出异常。

2、在reactive流中，必须使用reactive操作，如map、flatmap等。对对象执行操作。

例如，假设您想查看一个对象列表，将一些数据从web加载到每个对象，并将其保存到数据库中（注意，这里我将使用mock db和webservice，但您可以将它们更改为真正的服务，示例的本质是处理器。这里我也使用Kotlin，这与Java类似）

//Mock repository
object ReactorDatabase {
    fun saveData(car: Car): Mono<Car> = Mono.just(Car(1, car.producer,car.type))
    fun loadData(): Flux<Vehicle> = Flux.just(Vehicle("Toyota"), Vehicle("Ford"))
}  

//Mock webservice
object ReactiveWebService {
    fun loadFromWeb(): Mono<String> = Mono.just("corolla").delayElement(Duration.ofMillis(100))
}

处理器功能：

//Load vehicles
//Map vehicles to cars
//Load car type somewhere from the web and zip with the car
//Add type to car
//Save cars into database
//Return with the saved entites
fun process(): Flux<Car>{
        return ReactorDatabase.loadData()
            .map { Mapper.vehicleToCar(it) }           
            .zipWith(ReactiveWebService.loadFromWeb())
            .map { (car, type) -> Mapper.carWithDetails(car, type) }
            .concatMap { ReactorDatabase.saveData(it) }
    }

正如您所看到的，所有方法都是从反应流调用的。当您使用reactive IO操作时，它总是返回一个发布服务器。您可以使用flatmap、concatmap...获取对象并在不阻塞的情况下将其推向流。此外，如果您的操作没有阻塞，那么您可以从map中调用类似映射器的方法。

对于reactor，您不能使用命令式编程风格（除了使用协式或类似的struffs）。

这里还有一个反应性的Spring云流的例子。