파일의 첫 번째 N 바이트를 Java로 InputStream으로 읽으십니까?

Question

내 인생에서 내가하려는 일과 일치하는 질문을 찾을 수 없었으므로 여기에 어떤 유스 케이스가 있는지 설명하겠습니다. 이미 이에 대한 답변을 다루는 주제를 알고 있다면 자유롭게 저에게 안내해주십시오. :)

파일을 Amazon S3에 주기적으로 (매 20 초마다) 업로드하는 코드가 있습니다. 이 파일은 다른 프로세스가 작성한 로그 파일이므로이 함수는 실제로 로그가 끝나는 컴퓨터에 직접 액세스 할 필요없이 준 실시간으로 누군가의 내용을 읽을 수 있도록 로그를 꼬리 끌기위한 수단입니다 .

최근까지는 업로드하기 위해 단순히 파일을 입력으로 사용하는 S3 PutObject 방법을 사용했습니다. 그러나 AWS SDK 1.9에서는 실제로 업로드 된 콘텐츠 크기가 업로드 시작시 약속 된 콘텐츠 길이보다 큰 경우 S3 클라이언트가 요청을 거부하기 때문에 더 이상 작동하지 않습니다. 이 메서드는 데이터 스트리밍을 시작하기 전에 파일 크기를 읽습니다. 따라서이 응용 프로그램의 특성상 파일의 크기와 스트림 끝 사이의 크기가 증가 할 가능성이 매우 높습니다. 즉, 파일 크기에 관계없이 N 바이트의 데이터 만 보낼 수 있도록해야합니다.

어떤 식 으로든 파일의 바이트를 해석 할 필요가 없으므로 인코딩에 신경 쓰지 않아도됩니다. 바이트 단위로 전송할 수 있습니다. 기본적으로, 내가 원하는 것은 N 번째 바이트까지 파일을 읽을 수있는 간단한 방법입니다. 그런 다음 파일에 더 많은 데이터가있는 경우에도 읽기를 종료합니다. 즉, 특정 지점의 스트림에 EOF를 삽입하십시오. 예를 들어 업로드를 시작하면 파일 크기가 10000 바이트이지만 업로드 중에 12000 바이트로 증가하는 경우 업로드를 중지하고 싶습니다. 그 크기의 변화에 ​​관계없이 10000 바이트. (이후의 업로드에서 12000 바이트 이상을 업로드합니다.)

미리 만들어진 방법을 찾지 못했습니다. 지금까지 발견 한 것이 IOUtils.copyLarge 인 것으로 나타났습니다. InputStream, OutputStream, offset, length). 최대 길이 바이트를 제공된 OutputStream에 복사하도록 지시 할 수 있습니다. 그러나 copyLarge는 PutObject (아마도 InputStream에서 read() 형식을 호출하는 것처럼)를 차단하는 방법이기 때문에 전혀 작동하지 않을 것으로 보인다.

이렇게 할 수있는 메서드 나 기본 스트림을 찾지 못해서 얼마나 많은 바이트가 읽혔는지 직접 모니터하는 자체 구현을 작성해야한다고 생각하게 만듭니다. 그러면 아마도 일괄 처리 당 읽은 바이트 수가 버퍼 크기 또는 읽혀질 나머지 바이트 중 적은 수의 BufferedInputStream처럼 작동합니다. (예를 들어 3000 바이트의 버퍼 크기로, 각각 3000 바이트에서 3 개의 배치를 수행하고 1000 바이트 + EOF의 배치가 뒤따라야합니다.)

누구든지이 작업을 수행하는 더 좋은 방법을 알고 있습니까? 감사. 업로드 동안

이

(1) 나는 파일을 잠글 수 :

편집 그냥 난 이미 이상적입니다 어느 것도 몇 가지 대안, 알고 있어요 명확히한다. 이렇게하면 파일을 쓰는 프로세스에서 데이터가 손실되거나 작동 상 문제가 발생할 수 있습니다.

(2) 업로드하기 전에 파일의 로컬 복사본을 만들 수 있습니다. 이것은 매우 비효율적 일 수 있으며 많은 불필요한 디스크 공간을 차지할 수 있습니다 (이 파일은 수 기가 바이트 범위로 커질 수 있으며 실행중인 시스템은 디스크 공간이 부족할 수 있습니다).

편집 2 : 내 최종 솔루션, 동료의 제안에 근거은 다음과 같습니다

private void uploadLogFile(final File logFile) { 
    if (logFile.exists()) { 
     long byteLength = logFile.length(); 
     try (
      FileInputStream fileStream = new FileInputStream(logFile); 
      InputStream limitStream = ByteStreams.limit(fileStream, byteLength); 
     ) { 
      ObjectMetadata md = new ObjectMetadata(); 
      md.setContentLength(byteLength); 
      // Set other metadata as appropriate. 
      PutObjectRequest req = new PutObjectRequest(bucket, key, limitStream, md); 
      s3Client.putObject(req); 
     } // plus exception handling 
    } 
} 

LimitInputStream 내 동료, 그것은되지되었다는 것을 분명히 인식하지 무엇을 제안했다. ByteStreams.limit는 현재 구아바 대체품이며, 원하는대로 처리합니다. 고마워, 모두들.

Answer 1

전체 응답 찢어 & 교체 : 그것은 끝 데이터 신호 전에 제공 할 것입니다 바이트 수를 모자로서이

이 InputStream 등을 포장 비교적 간단합니다. FilterInputStream은이 일반적인 종류의 직업을 목표로하지만이 특정 직업에 대해 거의 모든 방법을 오버라이드해야하므로 그저 방해가됩니다. 왜 차단 I/O를 수행하는 문제가

import java.io.IOException; 
import java.io.InputStream; 

/** 
* An {@code InputStream} wrapper that provides up to a maximum number of 
* bytes from the underlying stream. Does not support mark/reset, even 
* when the wrapped stream does, and does not perform any buffering. 
*/ 
public class BoundedInputStream extends InputStream { 

    /** This stream's underlying @{code InputStream} */ 
    private final InputStream data; 

    /** The maximum number of bytes still available from this stream */ 
    private long bytesRemaining; 

    /** 
    * Initializes a new {@code BoundedInputStream} with the specified 
    * underlying stream and byte limit 
    * @param data the @{code InputStream} serving as the source of this 
    *  one's data 
    * @param maxBytes the maximum number of bytes this stream will deliver 
    *  before signaling end-of-data 
    */ 
    public BoundedInputStream(InputStream data, long maxBytes) { 
     this.data = data; 
     bytesRemaining = Math.max(maxBytes, 0); 
    } 

    @Override 
    public int available() throws IOException { 
     return (int) Math.min(data.available(), bytesRemaining); 
    } 

    @Override 
    public void close() throws IOException { 
     data.close(); 
    } 

    @Override 
    public synchronized void mark(int limit) { 
     // does nothing 
    } 

    @Override 
    public boolean markSupported() { 
     return false; 
    } 

    @Override 
    public int read(byte[] buf, int off, int len) throws IOException { 
     if (bytesRemaining > 0) { 
      int nRead = data.read(
        buf, off, (int) Math.min(len, bytesRemaining)); 

      bytesRemaining -= nRead; 

      return nRead; 
     } else { 
      return -1; 
     } 
    } 

    @Override 
    public int read(byte[] buf) throws IOException { 
     return this.read(buf, 0, buf.length); 
    } 

    @Override 
    public synchronized void reset() throws IOException { 
     throw new IOException("reset() not supported"); 
    } 

    @Override 
    public long skip(long n) throws IOException { 
     long skipped = data.skip(Math.min(n, bytesRemaining)); 

     bytesRemaining -= skipped; 

     return skipped; 
    } 

    @Override 
    public int read() throws IOException { 
     if (bytesRemaining > 0) { 
      int c = data.read(); 

      if (c >= 0) { 
       bytesRemaining -= 1; 
      } 

      return c; 
     } else { 
      return -1; 
     } 
    } 
}