为什么没有调用 Flux 发布者 doOnComplete 方法？

Question

我尝试在使用 Flux 发布者执行另一个方法后执行一个方法，但从未调用方法 doOnComplete。 代码如下：

public class Client implements Serializable {
    private Long id;
    private String category;
    // other properties, getters and setters
}

interface ClientRepository extends JpaRepository<Client,Long> {
    List<Client> findAllByCategory(String category);
    @Transactional
    void deleteByCategory(String category);
}

class ClientResponse {
    private Long status;
    private String message;
}

@Component
class ClientService {
    @Autowired
    ClientRepository clientRepository;
    @Autowired
    WebClient webClient;

  public Mono<ClientResponse> deleteRemoteClient(Long idClient) {
    return webClient.post()
      .uri("/api/remoteClient/{idClient}",idClient)
      .retrieve()
      .bodyToMono(ClientResponse.class)
      .doOnSuccess(ok -> System.out.println(
                          "Delete success for client= " + idClient))
      .doOnError(err -> System.out.println(
                          "Delete failed for client= " + idClient + ", err =" + err));
  }

  /**
   * Get All clients by category, then delete them remotely one by one.
   * When everything goes well, delete all clients locally in one shot by category
   **/
  public Flux<ClientResponse> deleteLocalClientByCategory(String category) {
    return Flux.fromStream(clientRepository.findAllByCategory(category).stream())
      .flatMap(client -> deleteRemoteClient(client.getId()))
      .doOnComplete(() -> clientRepository.deleteByCategory(category));
  }
}

@Component
class ClientHandler {
    @Autowired 
    ClientService service;

    public Mono<ClientResponse> deleteByCatgeory(ServerRequest request) {
        return service.deleteLocalClientByCategory(
                   Long.parseLong(request.queryParam("category").get()))
               .publishNext()
               .flatMap(response -> ServerResponse.ok().build());
    }
}

正如我之前提到的，方法deleteRemoteClient(client.getId()) 但不是clientRepository.deleteByCategory(category)。

Answer 1

当您调用publishNext 时，您会在第一个元素之后取消您的助焊剂。参见插图here。

所以你的助焊剂永远不会完成。您应该使用doOnCancel 或doFinally。

Answer 2

因此，为此目的使用.doOnComplite 不是一个好主意，请尝试.then 运算符，例如：

  ...
  .flatMap(client -> deleteRemoteClient(client.getId()))
  .then(clientRepository.deleteByCategory(category));