PipedOutput/InputStream 使用 spring 集成上传到 sftp答案

【问题标题】：PipedOutput/InputStream upload to sftp using spring integrationPipedOutput/InputStream 使用 spring 集成上传到 sftp
【发布时间】：2021-07-30 13:05:18
【问题描述】：

我想分几次将数据流式传输到 sftp 服务器。我正在使用 Spring Boot 集成。我已经像这样设置了 SftpRemoteFileTemplate

  @Autowired private SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory;

  @Bean
  public SftpRemoteFileTemplate sftpRemoteFileTemplate() {
    final SftpRemoteFileTemplate template = new SftpRemoteFileTemplate(sftpSessionFactory);

    template.setRemoteDirectoryExpression(
            new LiteralExpression(contactSenderSftpProperties.getSftpSessionProperties().getBaseSftpPath()));

    template.setTemporaryFileSuffix(".tmp");

    return template;
  }

但我的目标文件没有附加，而是被我要发送数据的临时文件的最新内容覆盖。

我的作者是这样的

  public void write(List<? extends Item> items) throws Exception {
    log.debug("Write {}", items);

    final int timeoutSeconds = 60;

    try (PipedInputStream pipedInputStream = new PipedInputStream()) {
      log.debug("Preparing to write...");
      final CountDownLatch countDownLatch = new CountDownLatch(1);

      writerClient.write(items, pipedInputStream, countDownLatch);

      if (!countDownLatch.await(timeoutSeconds, TimeUnit.SECONDS)) {
        throw new TimeoutException("Operation stream not connected");
      }

      sftpRemoteFileTemplate.send(
              MessageBuilder.withPayload(pipedInputStream).setHeader(FileHeaders.FILENAME, "contacts.csv").build());

    }
  }

方法 WriterClient#write

  @Async("writerThreadPoolTaskExecutor")
  public void write(List<? extends Item> items, PipedInputStream pipedInputStream, CountDownLatch countDownLatch) throws IOException {
    try(final PipedOutputStream pipedOutputStream = new PipedOutputStream(pipedInputStream)){
      countDownLatch.countDown();
      csvSerializer.serialize(pipedOutputStream, items.stream());
    }
  }

和 writerThreadPoolTaskExecutor

  @Bean
  public ThreadPoolTaskExecutor writerThreadPoolTaskExecutor(TaskExecutorBuilder taskExecutorBuilder) {
    return taskExecutorBuilder
            .corePoolSize(properties.getWriterThreadPoolCorePoolSize())
            .maxPoolSize(properties.getWriterThreadPoolMaxPoolSize())
            .queueCapacity(properties.getWriterThreadPoolQueueCapacity())
            .threadNamePrefix("writer-task-thread")
            .build();
  }

简而言之，我想编写许多小的临时文件并将它们合并到一个包含所有数据的文件中。我不太确定 PipedInput/OutputStream。当没有更多数据要写入时，是否可以多次附加 PipedOutputStream 并将 PipedInputStream 上传一次到 sftp ？但是问题来了，我怎么知道是否所有的日期都写了？

【问题讨论】：

sftp 确实有附加功能，所以理论上应该是可行的
可以，但不允许附加临时文件。
我可以在开头创建一个文件，附加到它，然后将其移动到目标目录。
是的； RemoteFileTemplate 有一个 append() 方法。
我决定不使用临时文件名功能。我的解决方案是将（每次调用追加）文件写入 sftp 服务器上的临时目录。当没有更多数据上传移动文件到目标目录时。

标签： java spring-boot spring-integration

【解决方案2】：

在RemoteFileTemplate 上使用append() 方法。

    /**
     * Send a file to a remote server, based on information in a message, appending.
     *
     * @param message The message.
     * @return The remote path, or null if no local file was found.
     * @since 4.1
     */
    String append(Message<?> message);

实现看起来像这样

    @Override
    public String append(Message<?> message, String subDirectory) {
        return send(message, subDirectory, FileExistsMode.APPEND);
    }

【讨论】：