기본 콘텐츠로 건너뛰기

라벨이 kafka-producer인 게시물 표시

[코드로 보는 카프카] Producer: BufferPool

 카프카 프로듀서는 메시지를 전송할 때 ByteBuffer를 사용한다. ByteBuffer는 생성하는 쓰레드에서 큰 메모리 단위를 생성하거나 여러 버퍼에 할당된 메모리를 해제할 때 쓰레드는 기아가 되거나 데드락이 될 수 있다. 이런 문제 때문에 프로듀서는 BufferPool을 사용한다.  이 BufferPool은 충분한 메모리가 확보 될 때 까지 쓰레드를 기다리게 할 수 있고, 이미 생성된 ByteBuffer를 재사용 할 수 있으며, 제한된 메모리로 동작할 수 있게 한다. 실제 BufferPool.java 는 메트릭 관련 코드도 있고 다른 변수들이 있어 예시보다는 쬐~금 복잡하지만 allocate와 deallocate 부분만 간단히 구현해보고 메모리를 제한하는 방법과 쓰레드를 처리하는 동작 방식에 대해 알아 두기로 한다. 코드를 보기 앞서 ByteBuffer.allocate와 ReentrantLock에 대해 먼저 알아보자. ByteBuffer.allocate  vs ByteBuffer.allocateDirect BufferPool은 allocate를 사용한다. 왜 allocateDirect를 사용하지 않는지는 여기 에 설명 되어 있는데, 간단히 요약하면 아래 정도의 내용이 된다. 생명주기가 짧거나 자주 사용되지 않는 객체에는 다이렉트 버퍼를 사용하지 않아야 한다. 왜냐하면, 다이렉트 버퍼는 OS 종속적인 네이티브 코드를 사용하기 때문에 힙기반 버퍼보다 생성과 메모리 반환 비용이 높고 가비지 컬렉터의 영역 밖이라 메모리 누수가 있을 수 있다. 용량이 큰 다이렉트 버퍼를 빈번하게 할당하면 OutofMemorryError가 생길 수 있다. 그리고, FileChannel 에서 non-direct Buffer와 direct Buffer 속도비교 ( FileChannel and non-direct buffer vs. FileChannel and direct buffer , 중국어! 코드와 그림만...

카프카 프로듀서(Producer) 0.9.0 설정

프로듀서 설정은 config/producer.properties 파일에서.. bootstrap.servers  호스트/포트 쌍으로 된 리스트로 카프카 초기에 클러스터에 연결할 때 사용한다. 이 리스트는 단순히 처음에 전체 서버를 찾는데 사용되는 호스트 리스트다 - host1:post1,host2:port2.. 형식 전체 클러스터 멤버(클러스터 멤버는 동적으로 바뀐다)를 찾는 초기 커넥션으로 사용하기 때문에, 모든 서버 리스트를 포함할 필요는 없다. (서버가 다운되었을 경우를 대비해 서버를 하나이상 적으면 된다) key.serializer, value.serializer 기본 org.apache.kafka.common.serialization.ByteArraySerializer 전달받은 byte[]를 그대로 리턴한다. buffer.memory  프로듀서가 서버로 보낼 레코드를 버퍼링 할 때 사용할 수 있는 전체 메모리의 바이트수 . 레코드가 서버로 전달 될 수 있는 것 보다 더 빨리 보내지면 버퍼는 소진되고 프로듀서는 max.block.ms 동안 레코드를 보내지 않고 블럭 한다. 블럭을 피하려면 block.on.buffer.full(아래참고) 을 false로 두면 된다.  이 세팅은 대략 프로듀서가 사용할 메모리 이지만, 프로듀서가 사용하는 메모리가 버퍼링만 있는게 아니기 때문에 설정에 딱떨어 지지는 않는다. 처리중인 요청 때문에 유지 해야하는 메모리 뿐만 아니라 압축 설정이 켜있으면 압축하는데 부가적인 메모리가 사용된다. 기본값은 약 33MB acks  acks=0: acks를 0으로 설정하면, 서버에서 수신확인(acknowledment)이 오길 기다리지 않는다 . 소켓 버퍼에 레코드가 즉시 추가되고 보내진것으로 간주된다. 이 경우에는 서버가 레코드를 수신했다는 보장이 없고, retries 설정도 먹지 않는다 (클라이언트는 통상 실패를 알지 못하기 때문에 retry를 할 수 ...