上传所有文件后如何回调？

Question

我需要从远程位置读取所有文件并将其发送到另一个服务，如果发送成功，则删除所有文件。我的代码对单个文件运行良好，但如果我想循环读取所有文件，则代码不会被执行。

请找到如下代码。在 RemoteFileReadImpl 类中，我正在尝试读取不起作用的循环文件。在 WebClientUtil 类中，我将文件发送到另一个服务。返回成功响应后，我想重命名已读取的文件。

package com.remotefileread.serviceImpl;

import java.io.IOException;

import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;

import reactor.core.publisher.Mono;

public class WebClientUtil{

    WebClient webClient = WebClient.create("http://localhost:9091");

    public Mono<HttpStatus> ftpFileSend(MultipartFile fileData) {

        MultiValueMap<String,Object> body=new LinkedMultiValueMap<String,Object>();

        try {
            body.add("file", fileData.getBytes());
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }


        return webClient
                .post()
                .uri("/storeFileData")
                .contentType(MediaType.MULTIPART_FORM_DATA)
                .body(BodyInserters.fromMultipartData(body))
                .exchange()
                .map(response -> {
                    
                    return response.statusCode();
                });
    }
    
}

    package com.remotefileread.serviceImpl;
    
    import java.io.File;
    import java.io.FileInputStream;
    import java.nio.file.Files;
    import java.util.Base64;
    
    import org.apache.commons.io.IOUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.http.HttpStatus;
    import org.springframework.http.MediaType;
    import org.springframework.mock.web.MockMultipartFile;
    import org.springframework.stereotype.Service;
    import org.springframework.web.multipart.MultipartFile;
    import org.springframework.web.reactive.function.server.ServerResponse;
    
    import com.remotefileread.model.SendFileData;
    import com.remotefileread.service.RemoteFileRead;
    
    import reactor.core.publisher.Mono;
    
    @Service
    public class RemoteFileReadImpl implements RemoteFileRead{
    
        @Autowired
        WebClientUtil webClientUtil;
    
        public Mono<ServerResponse> ftpFileRead() { 
            File directoryPath = new File("\\\\localhost\\SharedFolder\\csv_container");
            File files[] = directoryPath.listFiles();
            try {
                for(File csvFile : files) {
                    SendFileData fileData=new SendFileData();   
                    byte[] content = Files.readAllBytes(csvFile.toPath());
    
                    fileData.setFilename(csvFile.getName());
                    fileData.setFileContent(Base64.getEncoder().encodeToString(content));
                    fileData.setCustomerName("Cust");
                    FileInputStream input = new FileInputStream(csvFile);
                    MultipartFile multipartFile = new MockMultipartFile("file",
                            csvFile.getName(),"text/plain",IOUtils.toByteArray(input));
                    input.close();
    
                    Mono<HttpStatus> monoStatus = webClientUtil.ftpFileSend(multipartFile);
    
                    monoStatus
                    .doOnSuccess( httpStatus ->
                    {
                        System.out.println("Http Status:" + httpStatus);
                        
                    })
                    .doOnError(error -> 
                    {
                        System.out.println("Http Status:" + error);
                        
                    });
                }
                return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN).bodyValue("OK");
            }
            catch(Exception ex) {
                return ServerResponse.badRequest().contentType(MediaType.TEXT_PLAIN).bodyValue("Error Message: " + ex.getMessage());
            }
        }
    }

Answer 1

关于如何让你的问题变得更好的一些提示：

• 您提到了当每个文件成功时您希望发生什么，但如果一个文件失败则没有。它应该继续到下一个文件，还是停止并返回服务器错误？
• 说明它目前的表现，以及它与您的期望/愿望有何不同。
• 突出显示代码中不起作用的点。

如果您使用的是响应式，那么您通常不会像现在这样循环。您可能希望创建一个文件流以进行处理，例如 Flux:

Flux.fromArray(directoryPath.listFiles())

接下来，您将操作添加到该流的末尾，慢慢构建一个必须如何处理流以产生特定结果的“计划”。上面的Flux 准备将文件一个接一个地生成给订阅者。请注意，在您的代码中，没有任何内容订阅过 Mono，因此不会启动任何内容。

WebClient 也返回一个新流。 flatMap 函数允许将流中的元素映射到其他流，然后在原始流中元素的位置进行展平。在您的情况下，网络请求只是返回 Mono。

记住这一点，将创建 MultipartFile 的代码重构为另一个方法 createMultipart，并使用平面地图通过您的 util 类发出 Web 请求，我们将拥有以下流：

Flux<HttpStatus> statusStream = 
      Flux.fromArray(directoryPath.listFiles())
          .map(this::createMultipart)
          .flatMap(webClientUtil::ftpFileSend)

现在我们来谈谈WebClient的用法。

请注意，使用exchange()（现已弃用）意味着您需要确保使用响应数据，否则可能会导致内存泄漏。因此exchangeToMono() 或retrieve() 通常更好

如果您使用retrieve()，不成功的响应将自动引发异常，这将导致流中出现“错误”信号，从而停止处理任何进一步的文件。

所以总的来说，你可以有这样的实现：

    public Mono<ResponseEntity<Void>> ftpFileSend(MultipartFile fileData) {

        MultiValueMap<String,Object> body=new LinkedMultiValueMap<String,Object>();

        try {
            body.add("file", fileData.getBytes());
        } catch (IOException e) {
            return Mono.error(e);   // <-- note how to create an error signal
        }


        return webClient
                .post()
                .uri("/storeFileData")
                .contentType(MediaType.MULTIPART_FORM_DATA)
                .body(BodyInserters.fromMultipartData(body))
                .retrieve()
                .toBodilessEntity()
    }

...

    public Mono<ServerResponse> ftpFileRead() { 
    
        return Flux.fromArray(directoryPath.listFiles())
                   .flatMap(this::sendAndRename)
                   .onErrorContinue((ex, file) -> log("failed to process: " + file)) // <-- skips the erroring item and continues
                   .then(Mono.just(
                       ServerResponse.ok()
                             .contentType(MediaType.TEXT_PLAIN)
                             .bodyValue("OK"))
                   .onErrorResume(ex -> Mono.just(
                       ServerResponse.badRequest()
                             .contentType(MediaType.TEXT_PLAIN)
                             .bodyValue("Error Message:" + ex.getMessage())); // <-- not really a need if we just skip.
    }

    public Mono<Void> sendAndRename(final File file) {
        MultipartFile multipart = createMultipart(file);
        return webClientUtil.ftpFileSend(multipart)
                            .then(() -> renameDoneFile(file));
    }

这里发送所有文件。如果在发送或重命名时发生错误，则会记录该错误，跳过文件并继续处理下一个文件。