我有一个 Spring Webflux 反应式服务,它接收 DTO 并将其插入到多个表中。 有时我们可能需要根据传入的 DTO 跳过插入某些表。
这些是要求:
问题:-
public Mono<ServerResponse> createClientProfile(ServerRequest request) {
return secContext.retrieveUser().flatMap(usr -> {
return request.bodyToMono(ClientDto.class).flatMap(client -> {
return toNewClient(client, usr).flatMap(clientRepository::save).flatMap(clientRes -> {
return toNewClientReferral(clientRes.getClientId(), client.getDiscount(), usr)
.flatMap(clientReferralRepository::save).flatMap(clientReferralRes -> {
return toNewClientSyContact(clientRes.getClientId(), client.getSecondary(), usr)
.flatMap(clientSyContactRepository::save).flatMap(clientSyContactRes -> {
return clientPhoneRepository
.saveAll(toNewClientPhone(clientRes.getClientId(), client.getPhones(), usr))
.collectList().flatMap(phoneRes -> {
return ServerResponse
.created(URI.create(String.format(CLIENT_URI_FORMAT,
clientRes.getClientId())))
.contentType(APPLICATION_JSON).build();
});
});
});
});
});
});
}
private Mono<Referral> toNewClientReferral(final long clientId, final Discount dto) {
Referral referral = Referral.of(clientId,
dto.getName(), dto.getType(), dto.getAmount(), dto.getStatus());
return Mono.just(referral);
}
client.getDiscount() 可以为 null,
client.getSecondary() 可以为 null, client.getPhones() 可以为空。
我用 3 种不同的方法分离了流程。
public void createSyContact(ServerRequest request, long clientId) {
secContext.retrieveUser().flatMap(usr -> {
return request.bodyToMono(ClientDto.class).flatMap(client -> {
if (client.getSecondary() != null) {
return toNewClientSyContact(clientId, client.getSecondary(), usr)
.flatMap(clientSyContactRepository::save).flatMap(clientRes -> {
return Mono.just(clientRes.getClientId());
});
} else {
return Mono.empty();
}
});
});
}
public void createReferral(ServerRequest request, long clientId) {
secContext.retrieveUser().flatMap(usr -> {
return request.bodyToMono(ClientDto.class).flatMap(client -> {
if (client.getDiscount() != null) {
return toNewClientReferral(clientId, client.getDiscount(), usr)
.flatMap(clientReferralRepository::save).flatMap(clientRes -> {
return Mono.just(clientRes.getClientId());
});
} else {
return Mono.empty();
}
});
});
}
public Mono<Long> createClientWithPhones(ServerRequest request) {
return secContext.retrieveUser().flatMap(usr -> {
return request.bodyToMono(ClientDto.class).flatMap(client -> {
return toNewClient(client, usr).flatMap(clientRepository::save).flatMap(clientRes -> {
return clientPhoneRepository
.saveAll(toNewClientPhone(clientRes.getClientId(), client.getPhones(), usr)).collectList()
.flatMap(phoneRes -> {
return Mono.just(clientRes.getClientId());
});
});
});
});
}
这里,createClientWithPhones 是强制性的,所以没有 if 检查那里。但其他 2 个方法 createReferral 和 createSyContact 有 if 检查。需要先执行createClientWithPhones,它会返回clientId。此 clientId 应该在 createReferral 和 createSyContact 中使用。
public Mono<ServerResponse> createClientProfile(ServerRequest request) {
final List<Long> clinetIdList = new ArrayList<>();
createClientWithPhones(request).subscribe(result -> {
clinetIdList.add(result.longValue());
createSyContact(request, result.longValue());
createReferral(request, result.longValue());
});
return ServerResponse
.created(URI.create(String.format(CLIENT_URI_FORMAT,
clinetIdList.get(0))))
.contentType(APPLICATION_JSON).build();
}
这是处理这个问题的方法吗?
嗯,我认为人们对反应式库总体上没有很好的理解。我的意思是,人们通常会像 Java 8 流一样尝试进行函数式编程。当然,反应式库基于函数式编程,但我认为其目的是围绕阻塞 I/O 进行异步。考虑 WebFlux 项目的(当前)首页。
什么是反应式处理? 反应式处理是一种范例,使开发人员能够构建可以处理背压(流量控制)的非阻塞、异步应用程序。
所以,这是一种冗长的说法,我认为最好关注 I/O 发生的位置,而不是创建功能代码。如果您需要
if
语句,那么您需要 if
语句。与其试图弄清楚如何使用函数式编程执行 if
语句,不如尝试找出 I/O 发生的位置并以异步方式处理它。我喜欢使用的一个“技巧”是Mono::zip
或Flux::zip
。这些函数将许多 I/O 调用组合到一个发布者中以返回给客户端。因此,请考虑这个示例代码。
让我们制作一些反应式 r2dbc 函数:
Mono<Client> save(Client client) {
client.id = 1L;
System.out.println("Save client: " + client.id);
return Mono.just(client);
}
Mono<Phone> save(Phone phone) {
System.out.println("Save phone: " + phone.clientId);
return Mono.just(phone);
}
Mono<Referral> save(Referral referral) {
System.out.println("Save referral: " + referral.clientId);
return Mono.just(referral);
}
Mono<Contact> save(Contact contact) {
System.out.println("Save contact: " + contact.clientId);
return Mono.just(contact);
}
我们需要一些示例类来使用:
class DTO {
Client client;
List<Phone> phones;
Optional<Contact> contact;
Optional<Referral> referral;
}
class Client {
Long id;
}
class Contact {
Long clientId;
}
class Referral {
Long clientId;
}
class Phone {
Long clientId;
}
我们的输入可能是
Mono<DTO>
,因为这是请求应该提供的内容,因此我们的 Service
层需要以此开始并返回客户端 id 的 Mono<Long>
。
Mono<Long> doWork(Mono<DTO> monoDto) {
return monoDto.flatMap(dto->{
return save(dto.client).flatMap(client->{
List<Mono<?>> publishers = new ArrayList<>();
dto.phones.forEach(phone->{
phone.clientId = client.id;
publishers.add(save(phone));
});
if ( dto.contact.isPresent()) {
Contact c = dto.contact.get();
c.clientId = client.id;
publishers.add(save(c));
}
if ( dto.referral.isPresent()) {
Referral r = dto.referral.get();
r.clientId = client.id;
publishers.add(save(r));
}
if ( publishers.size() > 0 )
return Mono.zip(publishers, obs->client.id);
else
return Mono.just(client.id);
});
});
}
我使用以下示例代码运行了此代码:
@Override
public void run(ApplicationArguments args) throws Exception {
saveClient(new Client(), null, null, null).subscribe(System.out::println);
saveClient(new Client(), new Phone(), null, null).subscribe(System.out::println);
saveClient(new Client(), new Phone(), new Contact(), null).subscribe(System.out::println);
saveClient(new Client(), new Phone(), new Contact(), new Referral()).subscribe(System.out::println);
}
private Mono<Long> saveClient(Client client, Phone phone, Contact contact,
Referral referral) {
// TODO Auto-generated method stub
DTO dto = new DTO();
dto.client = client;
dto.phones = new ArrayList<>();
if ( phone != null ) dto.phones.add(phone);
dto.contact = Optional.ofNullable(contact);
dto.referral = Optional.ofNullable(referral);
return doWork(Mono.just(dto));
}
所以,这使用了
Mono.zip
技巧。保存的客户端是平面映射的,因此首先完成。然后为所有后续需要完成的保存创建一个单声道列表。这些单声道都是由 Mono.zip 函数异步执行的。 “combiner”函数对结果不执行任何操作,它只是返回客户端想要的 clientId。 Mono.zip 将所有 Mono 组合成一个 Mono 返回给客户端。从某种意义上说,这只是采用过程代码并将其包装在反应式库中,而不是过度关注函数式编程。如果业务“流程”未来发生变化,这很容易阅读和修改。
如果您喜欢的话,这是一个起点。我没有使用
Repository::saveAll
,所以这可能是一个改进。
确保所有
Flux
和 Mono
发布商都链接在一起非常重要。在你的最后一个例子中,你似乎放弃了它们。仅仅创建它们是不够的,它们都必须以某种方式返回给客户端。另外,您的代码有一个 subscribe
调用,这是一个禁忌。只有客户端应该订阅。我认为你应该在那里使用 map
。
编辑:修复了一个错误。仔细检查你的代码。
编辑二:我注意到我在类中使用了“可选”作为参数。这是一种反模式。正确的方法是使用类型作为参数,并将其包装在特殊 getter 中的“Optional.of”中。
可以在
flatMap
中完成简单的 if 语句,然后执行操作。
public Mono<String> foobar() {
return Mono.just("foo").flatMap(value -> {
if(value != null)
return Mono.just("Has value");
else
return Mono.empty();
}
}
foobar()
.switchIfEmpty(Mono.just("Is empty"))
.subscribe(output -> System.out.println(output);
我喜欢采用的一种方法是使用
filter
和 switchIfEmpty
的组合。
public void write(String content) {
Mono.just(content)
.filter(String::isEmpty)
.switchIfEmpty(Mono.just("No content"))
.subscribe(System.out::println);
}
filter
采用谓词来测试提供的值。如果谓词计算结果为 false,则它将在没有值的情况下完成(返回 Mono.empty())。