@Async API 端点 Spring Webflux

Posted 2023-02-19

【中文标题】@Async API 端点 Spring Webflux

【英文标题】：@Async API endpoint Spring Webflux

【发布时间】：2021-01-17 03:11:22

【问题描述】：

我需要编写 Spring Webflux 端点（路由器功能）以将邮件发送到邮件收件人列表。 UI 将选择邮件收件人列表并将列表发送到我将要编写的 API。我希望在收到请求后立即以这样的方式实施端点，我应该向 UI 发送响应，说明正在发送电子邮件。发送响应后，我应该继续异步发送邮件工作。我不能像在 Spring MVC 中那样使用 @async 注释，因为它是响应式世界中的反模式。

由于我使用spring webflux开发API，我该如何发送repsonse。

我的代码中有以下结构。

Router.java

@Bean
public RouterFunction<ServerResponse> sendEmail() 
 return route(POST("/email").and(accept(APPLICATION_JSON)), handler::sendEmail);

Handler.java

@Autowired
EmailService emailService;

public Mono<ServerResponse> sendEmail(ServerRequest request) 
    Mono<PojoA> pojoAMono = request.bodyToMono(PojoA.class);
    return pojoAMono.flatMap(pojoA -> 
       return emailService.sendEmail(pojoA).flatMap(mailSent -> 
         return  ServerResponse
        .status(HttpStatus.OK)
        .contentType(MediaType.APPLICATION_JSON)
        .body("Mails are being sent", String.class));
       );
    );
    

【参考方案1】：

您可以直接将响应返回给调用者，然后在该流完成后运行电子邮件发送作为副作用。这可以通过在流完成后执行的doFinally 来完成。

所以您的代码可能如下所示：

public Mono<ServerResponse> sendEmail(ServerRequest request) 
    return request.bodyToMono(PojoA.class)
            .map(this::sendEmailSideEffect)
            .flatMap(pojoA -> ServerResponse
                    .status(HttpStatus.OK)
                    .contentType(MediaType.APPLICATION_JSON)
                    .body("Mails are being sent", String.class));


private Mono<PojoA> sendEmailSideEffect(PojoA pojoA) 
    return Mono.just(pojoA)
            .doFinally(signalType -> emailService.sendEmails(pojoA));

【讨论】：

我尝试了上面的代码，当我测试它时，它似乎仍然会等待 sendMailSideEffect 完成，然后响应调用者。

【参考方案2】：

您应该构建响应，然后处理数据（发送电子邮件）。

类似这样的：

@Bean
public RouterFunction<ServerResponse> sendEmail() 
    return route(POST("/test").and(accept(APPLICATION_JSON)), this::someMethod);


Mono<ServerResponse> someMethod(ServerRequest serverRequest) 
    return ServerResponse.ok().build()
            .doOnNext(r -> Mono.just("data") //doing some process like send email
                    .delayElement(Duration.ofSeconds(2))
                    .subscribeOn(Schedulers.parallel())
                    .log()
                    .subscribe());

出于测试目的，我延迟了您查看响应发送后的数据处理。