[Java] NIO 기반 입출력 및 네트워킹 - 파일 비동기 채널

NIO 기반 입출력 및 네트워킹은 여러 절로 구성되어 있습니다.

[Java] NIO 기반 입출력 및 네트워킹 - NIO, 파일 & 디렉토리

[Java] NIO 기반 입출력 및 네트워킹 - 버퍼(Buffer)

[Java] NIO 기반 입출력 및 네트워킹 - 파일 채널

[Java] NIO 기반 입출력 및 네트워킹 - 파일 비동기 채널

[Java] NIO 기반 입출력 및 네트워킹 - TCP 블로킹 채널

[Java] NIO 기반 입출력 및 네트워킹 - TCP 넌블로킹 채널

[Java] NIO 기반 입출력 및 네트워킹 - UDP 채널

---

5. 파일 비동기 채널

 FileChannel의 read()와 write() 메소드는 파일 입출력 작업동안 블로킹(blocking)됩니다.

 만약 UI 및 이벤트를 처리하는 스레드에서 이 메소드들을 호출하면 블로킹되는 동안에 UI 갱신이나 이벤트 처리를 할 수 없습니다.

 따라서 별도의 스레드를 생성해서 이 메소드들을 호출해야만 합니다. 만약 동시에 처리해야 할 파일 수가 많다면 스레드의 수도 증가하기 때문에 문제가 될 수도 있습니다. 

 그래서 자바 NIO는 불특정 다수의 파일 및 대용량 파일의 입출력 작업을 위해서 비동기 파일 채널(AsynchronousFileChannel)을 별도로 제공하고 있습니다.

 AsynchronousFileChannel의 특징은 파일의 데이터 입출력을 위해 read()와 write() 메소드를 호출하면 스레드 풀에게 작업 처리를 요청하고 이 메소드들을 즉시 리턴시킵니다.

 실질적인 입출력 작업 처리는 스레드 풀의 작업 스레드가 담당합니다. 작업 스레드가 파일 입출력을 완료하게 되면 콜백(callback) 메소드가 자동 호출되기 때문에 작업 완료 후 실행해야 할 코드가 있다면 콜백 메소드에 작성하면 됩니다.

 5.1 AsynchronousFileChannel 생성과 닫기

   AsynchronousFileChannel  은 두 가지 정적 메소드인 open()을 호출해서 얻을 수 있습니다.

  첫 번째 open() 메소드는 다음과 같이 파일의 Path 객체와 열기 옵션 값을 파라미터로 받습니다.

AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(
    Path file,
    OpenOption ...options
);
 

  이렇게 생성된  AsynchronousFileChannel은 내부적으로 생성되는 기본 스레드풀을 이용해서 스레드를 관리합니다.

  기본 스레드풀의 최대 스레드 개수는 개발자가 지정할 수 없기 때문에 다음과 같이 두 번째 open() 메소드로 AsynchronousFileChannel 을 생성할 수 있습니다.

AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(
    Path file,
    Set<? extends OpenOption> options,
    ExecutorService executor,
    FileAttribute<?> ... attrs
);
 

  file 파라미터는 파일의 Path 객체이고, options 파라미터는 열기 옵션 값들이 저장된 Set 객체입니다.

  executor 파라미터는 스레드풀인 ExecutorService 객체입니다. attrs 파라미터는 파일 생성 시 파일 속성 값이 될 FileAttribute를 나열하면 됩니다.

  예를 들어 "C:\r_temp\file.txt" 파일에 대한 AsynchronousFileChannel은 다음과 같이 생성할 수 있습니다.

ExecutorService executorService = Executors.newFixedThreadPool(
    Runtime.getRuntime().availableProcessors()
);
 
AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(
    Paths.get("c:/r_temp/file.txt"),
    EnumSet.of(StandardOpenOption.CREATE, StandardOpenOption.WRITE),
    executorService
);
 

  Runtime.getRuntime().availableProcessors() 는 CPU의 코어 수를 리턴합니다.

  쿼드 코어 CPU일 경우는 4를 리턴, 하이퍼 스레딩일 경우는 8을 리턴합니다.

  EnumSet.of() 메소드는 파라미터로 나열된 열거 상수를 Set 객체에 담아서 리턴합니다.

   AsynchronousFileChannel을 더 이상 사용하지 않을 경우에는 다음과 같이 close() 메소드를 호출해서 닫아줍니다.

fileChannel.close();

  

 5.2 파일 읽기와 쓰기

   AsynchronousFileChannel이 생성되었다면 read()와 write() 메소드를 이용해서 입출력을 할 수 있습니다.

read(ByteBuffer dst, long position, A attachment, CompletionHandler<Integer, A> handler);
 
write(ByteBuffer src, long position, A attachment, CompletionHandler<Integer, A> handler);
 

  이 메소드들을 호출하면 즉시 리턴되고, 스레드 풀의 스레드가 입출력 작업을 진행합니다. 

  dst와 src 파라미터는 읽거나 쓰기 위한 ByteBuffer이고, position 파라미터는 파일에서 읽을 위치이거나 쓸 위치입니다.

  파일의 첫 번째 바이트부터 읽거나 첫 번째 위치에 바이트를 쓰고 싶다면 position을 0으로 주면 됩니다.

  attachment 파라미터는 콜백 메소드로 전달할 첨부 객체입니다. 

  첨부 객체는 콜밸 메소드에서 결과값 외에 추가적인 정보를 얻고자 할 때 사용하는 객체를 말합니다. 

  만약 첨부 객체가 필요 없다면 null을 대입해도 됩니다.

  handler 파라미터는 CompletionHandler<Integer, A> 구현 객체를 지정합니다. Integer는 입출력 작업의 결과 타입으로, read() 와 write()가 읽거나 쓴 바이트 수입니다.

  A는 첨부 객체 타입으로 개발자가 CompletionHandler 구현 객체를 작성할 때 임의로 지정이 가능합니다. 만약 첨부 객체가 필요 없다면 A는 void가 됩니다.

  CompletionHandler<Integer, A> 구현 객체는 비동기 작업이 정상적으로 완료된 경우와 예외 발생으로 실패된 경우에 자동 콜백되는 다음 두 가지 메소드를 가지고 있어야 합니다.

리턴 타입	메소드명(파라미터)	설명
void	completed(Integer result, A attachment)	작업이 정상적으로 완료된 경우 콜백
void	failed(Throwable exc, A attachment)	예외 때문에 작업이 실패된 경우 콜백

  completed() 메소드의 result 파라미터는 작업 결과가 대입되는데, read() 와 write() 작업 결과는 읽거나 쓴 바이트 수입니다.

  attachment 파라미터는 read()와 write() 호출 시 제공된 첨부 객체입니다.

  failed() 메소드의 exc 파라미터는 작업 처리 도중 발생한 예외입니다. 주목할 점은 콜백 메소드를 실행하는 스레드는 read()와 write()를 호출한 스레드가 아니고 스레드풀의 작업 스레드라는 점입니다.

  다음은 CompletionHandler 구현 클래스를 작성하는 방법을 보여줍니다.

new CompletionHandler<Integer, A>() {
    @Override
    public void completed(Integer result, A attachment) { ... }
 
    @Override
    public void failed(Throwable exc, A attachment) { ... }
};
 

  다음은 예제는 AsynchronousFileChannel을 이용해서 비동기적으로 "C:\r_temp" 디렉토리에 file0.txt ~ file9.txt 까지 총 10개의 파일을 생성한 후 "Hello ~ !!" 라는 내용을 씁니다.

  그리고 비동기 작업이 완료되었을 때 사용된 바이트 수와 처리를 담당했던 스레드 이름을 콘솔에 출력합니다.

  * AsynchronousFileChannelWriteExam.java

 
package pathexam;
 
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.EnumSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class AsynchronousFileChannelWriteExam {
 
    public static void main(String[] args) throws Exception {
        // 스레드풀 생성
        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
 
        for (int i = 0; i < 10; i++) {
            Path path = Paths.get("c:/r_temp/file" + i + ".txt");
            Files.createDirectories(path.getParent());
 
            AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(path,
                    EnumSet.of(StandardOpenOption.CREATE, StandardOpenOption.WRITE), executorService);
 
            Charset charset = Charset.defaultCharset();
            ByteBuffer buffer = charset.encode("Hello ~ !!");
 
            class Attachment {
                Path path;
                AsynchronousFileChannel fileChannel;
            }
 
            Attachment attachment = new Attachment();
            attachment.path = path;
            attachment.fileChannel = fileChannel;
 
            CompletionHandler<Integer, Attachment> completionHandler = new CompletionHandler<Integer, Attachment>() {
 
                @Override
                public void completed(Integer result, Attachment attachment) {
                    System.out.println(attachment.path.getFileName() + " : " + result + " bytes written : "
                            + Thread.currentThread().getName());
                    try {
                        attachment.fileChannel.close();
                    } catch (IOException e) {
                    }
                }
 
                @Override
                public void failed(Throwable exc, Attachment attachment) {
                    exc.printStackTrace();
                    try {
                        attachment.fileChannel.close();
                    } catch (IOException e) {
                    }
                }
 
            };
 
            fileChannel.write(buffer, 0, attachment, completionHandler);
        }
 
        executorService.shutdown();
    }
 
}
 

  이 예제에서 주의할 점은 write() 메소드가 즉시 리턴되더라도 뒤에서는 작업 스레드가 파일 쓰기 작업을 하고 있기 때문에 바로 AsynchronousFileChannel을 닫으면 안 됩니다.

  작업이 정상적으로 완료되었거나, 실패일 경우 채널을 닫아야 하므로 completed()와 failed() 메소드 에서  AsynchronousFileChannel의 close()를 호출해야 합니다.

  다음 예제는 이전 예제에서 생성한 file0.txt ~ file9.txt를 읽고 콘솔에 출력합니다.

  * AsynchronousFileChannelReadExam.java

 
package pathexam;
 
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.EnumSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class AsynchronousFileChannelReadExam {
 
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
 
        for (int i = 0; i < 10; i++) {
            Path path = Paths.get("c:/r_temp/file" + i + ".txt");
 
            AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(path,
                    EnumSet.of(StandardOpenOption.READ), executorService);
 
            ByteBuffer buffer = ByteBuffer.allocate((int) fileChannel.size());
 
            class Attachment {
                Path path;
                AsynchronousFileChannel fileChannel;
                ByteBuffer byteBuffer;
 
                public Attachment(Path path, AsynchronousFileChannel fileChannel, ByteBuffer byteBuffer) {
                    this.path = path;
                    this.fileChannel = fileChannel;
                    this.byteBuffer = byteBuffer;
                }
            }
 
            Attachment attachment = new Attachment(path, fileChannel, buffer);
 
            CompletionHandler<Integer, Attachment> completionHander = new CompletionHandler<Integer, Attachment>() {
 
                @Override
                public void completed(Integer result, Attachment attachment) {
                    attachment.byteBuffer.flip();
 
                    Charset charset = Charset.defaultCharset();
                    String data = charset.decode(buffer).toString();
 
                    System.out.println(
                            attachment.path.getFileName() + " : " + data + " : " + Thread.currentThread().getName());
 
                    try {
                        fileChannel.close();
                    } catch (IOException e) {
 
                    }
                }
 
                @Override
                public void failed(Throwable exc, Attachment attachment) {
                    exc.printStackTrace();
                    try {
                        fileChannel.close();
                    } catch (IOException e) {
 
                    }
                }
            };
            
            fileChannel.read(buffer, 0, attachment, completionHander);
            
        }
        executorService.shutdown();
 
    }
 
}
 

  이 예제에서도 역시 read() 메소드가 즉시 리턴되더라도 뒤에서는 작업 스레드가 파일 읽기 작업을 하고 있기 때문에 바로 AsynchronousFileChannel을 닫으면 안 됩니다.

  작업이 정상적으로 완료되었거나, 실패일 경우 채널을 닫아야 하므로 completed() 와 failed() 메소드에서 fileChannel에 대한 close() 작업이 이루어져야 합니다.

 

* 이 포스트은 서적 '이것이 자바다' 를 참고하여 작성한 포스트입니다.