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

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

关于如何改进问题的一些提示：

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

如果您使用的是reactive，那么您通常不会像现在这样循环。您可能希望创建一个文件流进行处理，作为通量：

Flux.fromArray(directoryPath.listFiles())

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

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

因此，请记住，将创建多部分文件的代码重构为另一个方法，并使用平面映射通过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));
    }

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