[NIO] Chanel 부분 정리

- 채널은 네이티브IO 서비스를 이용하기 위해 만들어 졌으며 스트림과 비슷하다
- byteBuffer <-> 채널 <-> byteBuffer
- 메모리 맵 파일, 파일 락킹 같은 기능 이용 가능
- 채널 하나로 양방향 통신 가능
- 크게 파일채널, 소켓채널이 있다


public interface Channel extends Closeable{
  public boolean isOpen();
  public void close() throws IOException;
}

public interface InterruptibleChannel extends Channel{
   public void close() throws IOException;
}
- 비동기적으로 채널을 닫거나(close), 인터럽트(interrupt)할 수 있다.

public interface ReadableByteChannel extends Channel{
   public int read(ByteBufffer dst) throws IOException;
}
- 채널 내용 읽기
public interface WritableByteChannel extends Channel{
   public int write(ByteBuffer src) throws IOException;
}
- 채널에 쓰기

* byteChannel 은 read(), write() 가능

ReadableByteChannel src = Channels.newChannel(System.in);   WritableByteChannel dest = Channels.newChannel(System.out);      ByteBuffer buffer = ByteBuffer.allocateDirect(1024);   while (src.read(buffer) != -1) {    buffer.flip();    while (buffer.hasRemaining()) {     dest.write(buffer);    }    buffer.clear();   }

* ScatteringByteChannel, GatheringByteChannel
 -  반복문을 피하고 코드가 명료
 - 시스템 콜과 커널 영역에서 프로세스 영역으로 버퍼 복사를 줄여주거나 없애준다

ByteBuffer [] buf = ...; //단 한번의 시스템 콜을 수행한다. //만약, 동일한 물리적 메모리를 참조하지 않는 경우 //단 한번의 커널 영역에서 프로세스 영역으로의 버퍼 복사 scatteringByteChannel.read(buf);

FileInputStream fin = new FileInputStream("StreamTest.java");   ScatteringByteChannel channel = fin.getChannel();      ByteBuffer header = ByteBuffer.allocateDirect(100);   ByteBuffer body = ByteBuffer.allocateDirect(200);     ByteBuffer[] buffers = { header, body };        int readCount = (int) channel.read(buffers);   channel.close();   System.out.println("Read Count : " + readCount);      System.out.println("\n//--------------------------------------------------//\n");      header.flip();   body.flip();      byte[] b = new byte[100];   header.get(b);   System.out.println("Header : " + new String(b));      System.out.println("\n//--------------------------------------------------//\n");      byte[] bb = new byte[200];   body.get(bb);    System.out.println("Body : " + new String(bb));

- 순서대로 읽고 쓰여진다
- 통신 프로토콜을 만들어 사용할때, 프로토콜은 헤더라는 버퍼에, 전달 내용은 바디라는 버퍼에 넣어서 전달 할수 있다

* 파일 채널
1. 파일 채널은 항상 블록킹 모드며 비블록킹 모드로 설정할 수 없다.
2. 파일 채널 객체는 직접 만들 수 없다.
 - 이미 열려있는 파일 객체로부터 getChannel() 메소드를 호출하여 생성
3. 파일 채널 객체는 스레드에 안전하다
 - 어떤 스레드가 파일 크기 또는 파일채널의 포지션을 변경하는 부분을 수행하는 메소드를 호출하면
    다른 스레드들은 그 스레드가 해당 작업을 마무리할 때까지 기다렸다가 수행하도록 되어 있다.

ⓐ 기본속성
- 파일 채널은 파일 디스크립터와 일대일 관계를 맺는다
- long 형이다
- 해당 파일이 파일 디스크립터를 공유함으로 포지션을 변경하면 다른 인스턴스에도 모두 바로 반영

- truncate() :  파일이 주어진 값보다 크다면, 주어진 값의 크기로 잘라진다
- force(): 파일의 업데이트된 내요을 강제적으로 기억장치에 기입
              데이터 캐시를 하지 않는다.

ⓑ 파일 락킹
- 파일 락킹은 채널이 아닌 파일을 대상으로 하는 것
- 파일 락킹은 동일한 jvm 내부의 여러 스레드 사익 아닌, 외부 프로세스 사이에서 파일이 접근을 제어하기 위한 것이다.

//파일 위치 0부터 파일이 갖을 수 있는 최대 크기까지 락 설정, true 공유락, false 배타락  channel.lock(0L, Long.MAX_VALUE, false)  // 비블록킹 메소드 , 이미 락이된 상태라면 null 리턴  trylock()

- FileLock 클래스

* release() 메소드로  항상 해당 자원을 확실하게 닫을 수 있도록 코드내에서 보장해야한다. FileLock lock = fileChannel.lock(); try{ //어떤 처리 catch(Exception e){ //예외 핸들링... }finally{  //이곳에서 학실하게 중요 자원에 대한 처리를 마무리한다  lock.release(); }

FileChannel channel = null; try{   File file = new File("aaa.txt");   channel = new RandomAccessFile(file, "rw").getChannel();   FileLock lock = channel.lock(0, Long.MAX_VALUE, true);   try{    boolean isSharee = lock.isShared();   finally{     lock.release();   } }catch(Exception e){  e.printStackTrace(); }finally{  if(channel != null){    try{      channel.close();    }catch(IOException ex){    }  } }

* 메모리 맵핑
- map() 메소드의 호출을 통해 열려진 파일과 byteBuffer사이에 가상 메모리가 생성
  생성된 가상 메모리는 파일을 저장소로 사용하고 이 가상 메모리 영역은 MappedByteBuffer 객체가 래핑하게 된다
- MappedByteBuffer 에 읽거나 쓰면 곧바로 파일에 반영
- 자바 IO, 채널을 이용하는 것보다 효율적. 시스템 콜이 없으며 운영체제의 가상 메모리 시스템은 자동적으로 메모리 페이지들을 캐시한다. jvm으 힙을 사용하지 않고 시스템 메모리를 사용해서 캐시하기 때문에 한번 메모리 페이지가 정상적으로 만들어지면 그 데이터를 얻기 위해 다른 시스템 콜을 할 필요도 없이 하드웨어의 최고 속도로 데이터에 접근 할 수 있다.
- 큰 파일에 유용
- 파일의 전체를 메모리 매핑 할때

mapBuffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, fileChannel.size());

- 파일 매핑에서는 정해진 파일 크기만큼만 매핑되고 동적으로 매핑된 사이즈를 늘릴 수 없다.
- 메모리 매핑을 사용할 때에는 외부 프로세스가 해당 팡리에 읽기, 쓰기 이외의 동작을 하지 못하게 한다. 락을 사용한다
- 파일채널이 닫혀도 매핑은 해제되지 않고 MappedByteBuffer 가비지 컬렉트되어 객체 스스로가 제거되어야만 매핑이 깨진다
- jvm 힙이 아닌 다이렉트 버퍼로 소켓 패널 등에 read(),write() 메소드를 통해 효율적으로 데이터를 전송하기 위한 타겟으로 사용할 수 있다

- load() 메소드
: 가상 메모리가 만들어지는 시점에서는 단지 가상 메모리를 통해 그 파일에 접근할 준비가 된 것뿐이고 가상 메모리로 잡인 파일 부분들이항상 가상 메모리 생성 시점에 시스템 메로리로 로드된다고 보장할 수 없다. 따라서 캐시를 통한 빠른 접근을 위해서는 일단 해당 부분을 메모리로 로드해야한다.
MappedByteBuffer 를 사용할 때 운영체제의 시스템 메모리에 캐시되지 않는 부분을 로드하느라 걸리는 지연 시간을 없애고 빠르게 사요하기 위해서 load()메소드가 제공된다. 이 메소드는 운영체제의 시스템 메모리에 매핑된 파일 데이터를 로드할 수 있을 만큼 로드한다.

- force() 메소드
: 물리적 기억장치와의 동기화를 유지하기 위해 기억장치에 강제로 MappedByteBuffer 의 변경 사항을 저장

- ByteBufferPool
: 메모리와 파일 매핑을 통해 생성한 버퍼를 가지고 평소에는 일반적으로 성능이 뛰어난 메모리 버퍼를 사용하고 초기에 설정한 메모리 버퍼가 이미 모두 사용중일때에는 파일 매핑을 통해 생성한 가상 메모리 버퍼, 즉 MappedByteBuffer 를 사용하여 물리적 메모리의 한계를 극복할 수 있도록 만든 버퍼.

ByteBufferPool.java

import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.util.ArrayList; public class ByteBufferPool {  private static final int MEMORY_BLOCKSIZE = 1024;  private static final int FILE_BLOCKSIZE = 2048;    private final ArrayList memoryQueue = new ArrayList();  private final ArrayList fileQueue = new ArrayList();    private boolean isWait = false;    //할당할 메모리 크기, 맵핑할 파일 크기, 맵핑할 파일 지정  public ByteBufferPool(int memorySize, int fileSize, File file) throws IOException {   if (memorySize > 0)    //메모리 초기화    initMemoryBuffer(memorySize);   if (fileSize > 0)    //맵핑 파일 초기화    initFileBuffer(fileSize, file);  }    //메모리 초기화 (할달할 메모리 크기)  private void initMemoryBuffer(int size) {   //메모리 블럭 크기 단위로 크기를 구한다   int bufferCount = size / MEMORY_BLOCKSIZE;   size = bufferCount * MEMORY_BLOCKSIZE;   //다이렉트 메모리 할당   ByteBuffer directBuf = ByteBuffer.allocateDirect(size);   //할당된 메모리를 분할 (다이렉트버퍼, 블럭단위, 메모리 큐)   divideBuffer(directBuf, MEMORY_BLOCKSIZE, memoryQueue);  }    //맵핑파일 초기화(크기, 파일)  private void initFileBuffer(int size, File f) throws IOException {   //멥핑 파일 블럭 크기 단위로 크기를 재구성   int bufferCount = size / FILE_BLOCKSIZE;   size = bufferCount * FILE_BLOCKSIZE;   //맵핑 파일 생성   RandomAccessFile file = new RandomAccessFile(f, "rw");   try {    //맵핑 파일의 크기 설정    file.setLength(size);    //파일 전체를 파일 맵핑    ByteBuffer fileBuffer = file.getChannel().map(FileChannel.MapMode.READ_WRITE, 0L, size);    //맵핑할 크기 분할    divideBuffer(fileBuffer, FILE_BLOCKSIZE, fileQueue);   } finally {    file.close();   }  }  //사용할 단위 나누기  private void divideBuffer(ByteBuffer buf, int blockSize, ArrayList list) {   //버퍼 카운트 알아냄   int bufferCount = buf.capacity() / blockSize;   int position = 0;   //블럭 단위만큼의 카운트 만큼   for (int i = 0; i < bufferCount; i++) {    int max = position + blockSize;    //읽어들일 limit 설정    buf.limit(max);    //사용 단위 만큼 버퍼 자름    list.add(buf.slice());    //다음에 나눌 버퍼    position = max;    buf.position(position);   }  }    //메모리 버퍼 얻기  public ByteBuffer getMemoryBuffer() {   return getBuffer(memoryQueue, fileQueue);  }  //가상메모리 파일 얻기  public ByteBuffer getFileBuffer() {   return getBuffer(fileQueue, memoryQueue);  }  //실제로 메모리 얻는 부분  //메모리 버퍼를 얻고 없으면 가상메모리를 얻는다  //가상메모리를 얻고 없으면 메모리를 얻는다  private ByteBuffer getBuffer(ArrayList firstQueue, ArrayList secondQueue) {   //첫번째 메모리를 얻으려한다   ByteBuffer buffer = getBuffer(firstQueue, false);   //첫번재 메모리에서 사용할 버퍼가 없다면   if (buffer == null) {    //두번째 메모리를 얻으려한다    buffer = getBuffer(secondQueue, false);    //두번재 메모리도 없다면    if (buffer == null) {     //큐가 얻을때까지 대기하길 원한다면,     if (isWait)      buffer = getBuffer(firstQueue, true);     else      //대기하지 않고 바로 내부 메모리를 할당하고자 할때      buffer = ByteBuffer.allocate(MEMORY_BLOCKSIZE);    }   }   return buffer;  }  private ByteBuffer getBuffer(ArrayList queue, boolean wait) {   //큐가 다른 곳에서 동시에 사용되며 추가 및 삭제가 발생하기 때문에 동기화   synchronized (queue) {    //더이상 사용할 버퍼가 없다면    if (queue.isEmpty()) {     //큐가 얻을때까지 대기하길 원한다면,     if (wait) {      try {       //큐에 버퍼가 반환될때까지 기다린다       queue.wait();      } catch (InterruptedException e) {       return null;      }     //내부 메모리를 얻고자 할때     } else {      return null;     }    }    //사용될 버퍼를 리턴후 큐에서 삭제    return (ByteBuffer) queue.remove(0);   }  }  //사용한 버퍼를 반환 할때  public void putBuffer(ByteBuffer buffer) {   if (buffer.isDirect()) {    switch (buffer.capacity()) {     //직접 메모리라면     case MEMORY_BLOCKSIZE :      putBuffer(buffer, memoryQueue);      break;     //가상 메모리라면     case FILE_BLOCKSIZE :      putBuffer(buffer, fileQueue);      break;    }   }  }    //사용한 버퍼를 반환하는 구현부  private void putBuffer(ByteBuffer buffer, ArrayList queue) {   //사용된 버퍼의 속성 초기화   buffer.clear();   //동기화 필요   synchronized (queue) {    //사용한 버퍼를 다시 큐에 넣는다    queue.add(buffer);    //대기 중인 큐가 있으면 깨운다    queue.notify();   }  }    //사용된 버퍼가 반환 될때까지 대기할것인지 내부 메모리를 할당하여 사용할것인지 설정  public synchronized void setWait(boolean wait) { this.isWait = wait; }   public synchronized boolean isWait() { return isWait; } }

- 채널 간 직접 전송
: 파일채널에만 존재하는 메소드
  소켓채널은 사용할 수 없지만 파일의 내용을 transferTo()메소드를 이용하여 소켓으로 전달하거나
  소켓 큐에 저장된 내용을 transferFrom() 메소드를 이용하여 파일로 읽어올 수는 있다

- transferTo(long position,long count, WritableByteChannel target)
: 주어진 target 채널로 현재 파일채널이 position부터 count 만큼 파일로 읽기
: 전송되지 않는 경우
  1. 파일채널으 바이트 수가 지정된 position부터 시작되는 count보다 적은 경우
  2. 타겟 채널이 비블록 모드의 소켓으로 출력 버퍼의 비어있는 공간이 전송하려는 바이트의 양보다 적은 경우
    - 이런 경우 전송 가능한 만큼 전송하닥 현재까지 전송한 값을 리턴하고 메소드 실행 종료

-transferFrom()
 : 주어진 src 채널로 부터 최대 count만큼의 데이터를 읽어들여 파일채널에 지정된 position에 기입하려고 노력한다

- 이 메소드들은 파일채널의 포지션을 업데이트하지 않음
- 운영체제에서는 유저 영역을 통한 데이터의 전송 없이 파일 시스템 캐시로부터 데이터를 채널로 직접 전송하기때문에
   대량의 데이터를 전송해야 하는 경우에 큰 성능을 발휘

- 각 테스트 파일들

MyTimer.java

CopyByteBufferTest.java

CopyDirectByteBufferTest.java

CopyIOTest.java

CopyMapTest.java

CopyTransferFromTest.java

CopyTransferToTest.java

* 소켓 채널
- 기존의 소켓이나IO는 블록킹 된다는 단점이 있었다. 대신에 클라이언트당 쓰레드를 생성하여 사용하면 성능이 저하 됬다.

1. 비 블록킹 모드
 - 임계영역 안에서 다른 스레드에 영향을 주지 않고 잠시 소켓채널의 블록킹 모드를 변경후 다시 원상태로 돌려놓음
 - 소켓채널의 블록킹 모드를 변경하는 것을 막을 필요가 있을 때

Object lockObj = serverSocketChannel.blockingLock(); //임계영역 안에서 처리하므로 다른 스레드에 영향을 주지 않지만 //성능 저하를 가져온다 synchronized(lockObj){    //서버 소켓채널의 현재 모드를 저장한다    boolean preMode = serverSocketChannel.isBlocking();   //서버 소켗채널을 비블록킹 모드로 설정한다.    serverSocketChannel.configureBlocking(false);   //접속한 클라이언트를 처리한다.    Socket socket = serverSocketChannel.accept();    registeClient(socket);   //서버소켓채널을 처음 모드로 돌려놓는다.    serverSocketChannel.configureBlocking(preMode); }

 - 비블록킹 모드는 서버 쪽에서 사용 된다. 동시에 많은 소켓들을 쉽게 제어 할 수 있기 때문
 - p2p 경우 클라이언트쪽에서 필요

2. 서버소켓 채널
- 비블록킹 모드로 동작할 수있다.
- bind()가 없다

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();  ServerSocketChannel.configureBlocking(false);  ServerSocket serverSocket =  ServerSocketChannel.socket();  InetAddress ia = InetAddress.getLocalHost();   // InetAddress ia = InetAddress.getByName("192.168.0.2");  int port = 8080;  InetSocketAddress isa = new InetSocketAddress(ia, port);  serverSocket.bind(isa);

-  ServerSocketChannel.accept()는 SocketChannel을 리턴하고 ServerSocket은 Socket을 리턴한다.
-  ServerSocketChannel의 비블록킹 모드경우 접속을 시도하는 클라이언트가 없을 경우 null을 리턴

import java.net.InetSocketAddress; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; public class SSCAcceptExample {  public static void main(String[] args) throws Exception {   ServerSocketChannel ssc = ServerSocketChannel.open();   ssc.socket().bind(new InetSocketAddress(8080));   ssc.configureBlocking(false);   while (true) {    System.out.println("커넥션 연결을 위해 대기중..");        SocketChannel sc = ssc.accept();     if (sc == null) {     Thread.sleep(1000);    } else {     System.out.println(sc.socket().getRemoteSocketAddress() + " 가 연결을 시도했습니다.");    }   }  } }

3. 소켓 채널
- socket() 메소드를 호출해서 Socket 객체를 얻은 후, 이 소켓 객체의 connect(SocketAddress endpoint)메소드를 사용하여 연결
- open(SocketAddress remote) 를 이용하여 소켓채널을 생성함과 동시에 원격지의 서버로 연결

SocketChannel sc = SocketChannel.open(new InetSocketAddress("ip",port);  sc.configureBlocking(false); //위와 같은 방법 SocketCahnnel sc = SocketChannel.open(); sc.configureBlocking(false); sc.connect(new InetSocketAddress("ip", port)); sc.finishConnect(); //반드시 finishConnect()을 해야한다. 하지않으면, 소켓채널의 연결 시도는 대기상태로 있는다 //finishConnect()는 비블록킹에서만 유효하다. 서버에 연결을 시도하고 마무리한다

import java.net.InetAddress; import java.net.InetSocketAddress; import java.nio.channels.SocketChannel; public class SCConnectionTest {    private static int PORT = 8988;  public static void main(String[] args) throws Exception {   InetAddress ia = InetAddress.getByName("192.168.89.129");   InetSocketAddress isa = new InetSocketAddress(ia, PORT);      SocketChannel sc = SocketChannel.open();   sc.configureBlocking(false);   System.out.println("Is ConnectionPending 1 : " + sc.isConnectionPending());   sc.connect(isa);   System.out.println("Is ConnectionPending 2 : " + sc.isConnectionPending());   sc.finishConnect();   System.out.println("Is ConnectionPending 3 : " + sc.isConnectionPending());      // 또는 다음을 사용해도 같은 결과를 얻을 수 있음.. //  SocketChannel sc = SocketChannel.open(isa); //  sc.configureBlocking(false);      System.out.println("Is Connected : " + sc.isConnected());   System.out.println("Is Blocking Mode : " + sc.isBlocking());  } }

4. 데이터그램채널
- 블록킹 모드일 경우
: 일거들일 데이터가 있다면 버포로 읽어들이고 해당 패킷을 보낸 쪽의 주소에 리턴한다.
이때 패킷이 크기보다 파라미터로 주어진 버퍼에 저장할 크기가 더 작다면 버퍼에 저장 가능한 만큼 저장되고 나머지는 폐기
- 블록킹 모드인 경우
:  읽어들일 패킷이 없으면 즉시 null 리턴
- send()경우 버퍼 내용과 목적지 주소를 합쳐서 네트워크로 흘린다
원격지의 어떤 주소에 연결되어 있지 않고 보안관리자가 설치되어 있는 경우, 받거나 보내는 모든 패킷을 보안관리자의 checkAccept()메소드에 의해 인증된 주소(IP+PORT)인지 여부 확인하는데 이때 과부하를 피하기위해
connect(SocketAddress remote)가 사용된다. 이는 원격지 주소에 연결하지만 성능에 문제가 있는 경우만 사용
호출시 허용된 주소라면 이후에 이 데이터그램채널로 수신되거나 전송되는 모든 패킷은 설정된 주소에 대해서는 보안검사를 하지 않는다. 반대라면 Security Exception 발생
- 주소와 묶에서 보내기때문에 read(),wirte()를 사용할수 없고 접속이 안된상태에서는 receive(),send()를 사용
connet()메소드로 원격지와 연결된 상태여야 한다.

- 에코서버

import java.io.IOException; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.nio.ByteBuffer; import java.nio.channels.DatagramChannel; public class UDPEchoServer {  /**   * @param args   */    protected int port;    public UDPEchoServer(int port){   this.port = port;  }  public void execute()throws IOException, InterruptedException{   //비블록킹 모드로 데이터그램채널의 로컬 호스트 8080포트에 바인딩   DatagramChannel channel = DatagramChannel.open();   channel.socket().bind(new InetSocketAddress("localhost", port));   channel.configureBlocking(false);      ByteBuffer buffer = ByteBuffer.allocateDirect(1024);   while(true){    buffer.clear();    //패킷 수신    SocketAddress addr = channel.receive(buffer);    //도착된 패킷이 있다면 다시 보냄    if(addr != null){     System.out.println("패킷도착");     buffer.flip();     channel.send(buffer, addr);    }else    {     System.out.println("도착한 패킷이 없음");     Thread.sleep(5000);    }   }  }    public static void main(String[] args) throws IOException, InterruptedException {   // TODO Auto-generated method stub   UDPEchoServer server = new UDPEchoServer(8080);   server.execute();  } }

- 에코 클라이언트

import java.net.InetSocketAddress; import java.net.SocketAddress; import java.nio.ByteBuffer; import java.nio.channels.DatagramChannel; import java.util.Timer; import java.util.TimerTask; public class UDPEchoClient {  /**   * @param args   */    private static Timer timer = null;  public UDPEchoClient(int seconds) throws Exception{   timer = new Timer();   //2초마다 EchoClientTask 수행   timer.schedule(new EchoClientTask(), seconds*1000);   Thread.sleep(10000);   timer.cancel();  }    public static void main(String[] args) throws Exception {   // TODO Auto-generated method stub   new UDPEchoClient(2);  } class EchoClientTask extends TimerTask{  @Override  public void run() {   // TODO Auto-generated method stub   try{        //블록킹 모드의 데이터그램 채널 생성    DatagramChannel channel = DatagramChannel.open();    //에코 서버의 주소 생성 및 다이렉크 버퍼 생성    SocketAddress sa = new InetSocketAddress("localhost", 8080);    ByteBuffer buffer = ByteBuffer.allocateDirect(1024);    while(!Thread.interrupted()){     buffer.clear();     buffer.put("데이터그램채널 테스트...".getBytes());     buffer.flip();     //서버에 메시지 전송     channel.send(buffer, sa);     //버퍼 재사용     buffer.clear();     //서버로 메시지 수신     SocketAddress addr = channel.receive(buffer);     buffer.flip();     //받은 메시지 출력     byte [] bb = new byte[buffer.limit()];     buffer.get(bb, 0, buffer.limit());     String data = new String(bb);     System.out.println("Receive : "+data);          Thread.sleep(1000);    }   }catch(Exception e){    e.printStackTrace();   }  }   }   }

UDPEchoClient.java

UDPEchoServer.java