Windows환경에서 kafka클러스터 구축

Windows환경에서 kafka클러스터 구축

최초 zookeeper, kafka 단일로 구축하여 실행해보는 것은 아래의 내용을 참고 바라며

https://mr-spock.tistory.com/46

[#5 Apache Kafka 구축] Spring boot 기반 마이크로서비스 아키텍처(Microservices Architecture, MSA)

지난번 작성한 글은 zookeeper 클러스터 환경을 구축해보았다 해당 내용에 하게 된 동기나, 내용 등이 있으니 참고 바라며

https://mr-spock.tistory.com/58

Windows환경에서 zookeeper 클러스터(Ensemble, 앙상블) 구축

이이제 kafka클러스터 설정을 할 차례다. 실무위주로 바로 진행하겠다.

1. kafka복제 및 설정

지난번과 동일하게 일단 여러 개를 띄우려면 kafka소스가 여러 개만큼 복제를 해줘야 한다(서버를 다 별도로 두 려면 당연하고, 같은 호스트 내에서도 동일하다). 지금은 테스트 서버가 하나이기 때문에 호스트가 여러 개인 양 kafka소스를 3개 정도 복제하겠다. 아래의 구조와 폴더 내 구조를 참고하자

E:\kafka1
E:\kafka2
E:\kafka3

kafka 소스구조

이제 각 소스 별 /config/server.properties 파일을 아래와 같이 수정한다.

 

kafka1/config/server.properties

broker.id=1
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://localhost:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

log.dirs=E:\\kafka1\\logs
num.partitions=1
num.recovery.threads.per.data.dir=1

offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

zookeeper.connect=localhost:2181, localhost:2182, localhost:2183 
zookeeper.connection.timeout.ms=18000

group.initial.rebalance.delay.ms=0

kafka2/config/server.properties

 

broker.id=2
listeners=PLAINTEXT://:9093
advertised.listeners=PLAINTEXT://localhost:9093
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

log.dirs=E:\\kafka2\\logs
num.partitions=1
num.recovery.threads.per.data.dir=1

offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

zookeeper.connect=localhost:2181, localhost:2182, localhost:2183 
zookeeper.connection.timeout.ms=18000

group.initial.rebalance.delay.ms=0

kafka3/config/server.properties

 

broker.id=3
listeners=PLAINTEXT://:9094
advertised.listeners=PLAINTEXT://localhost:9094
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

log.dirs=E:\\kafka3\\logs
num.partitions=1
num.recovery.threads.per.data.dir=1

offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

zookeeper.connect=localhost:2181, localhost:2182, localhost:2183 
zookeeper.connection.timeout.ms=18000

group.initial.rebalance.delay.ms=0

• broker.id 부분은 클러스터 환경에서 해당 kafka가 유일하도록 지정한다.
• listeners=PLAINTEXT://:9094 
  advertised.listeners=PLAINTEXT://localhost:9094
  두 부분의 포트가 kafka의 기본 포트인데 같은 서버에 세팅하였으니, 겹치지 않게 1씩 더해줬다.
• zookeeper.connect=localhost:2181, localhost:2182, localhost:2183  이 부분은 3개의 kafka설정 동일하게 zookeeper클러스터 구성할 때 작성했던 내용을 적용한다.

 

두 번째로 각 소스 별 /config/zookeeper.properties 파일을 아래와 같이 수정한다.

 

kafka1/config/zookeeper.properties

dataDir=/tmp/zookeeper
clientPort=2181
maxClientCnxns=0
admin.enableServer=false

initLimit=5
syncLimit=2
server.1=localhost:2888:3888
server.2=localhost:2889:3889
server.3=localhost:2890:3890

kafka2/config/zookeeper.properties

dataDir=/tmp/zookeeper
clientPort=2182
maxClientCnxns=0
admin.enableServer=false

initLimit=5
syncLimit=2
server.1=localhost:2888:3888
server.2=localhost:2889:3889
server.3=localhost:2890:3890

kafka3/config/zookeeper.properties

dataDir=/tmp/zookeeper
clientPort=2183
maxClientCnxns=0
admin.enableServer=false

initLimit=5
syncLimit=2
server.1=localhost:2888:3888
server.2=localhost:2889:3889
server.3=localhost:2890:3890

 

이렇게 설정 한 후 각 kafka서버를 기동 한다. 주의할 점은 기존에 kafka를 운영했었다면, /logs폴더의 meta.properties파일을 삭제 후 기동해야 한다. 해당 파일에 broker.id가 지정되어 있어서 일치하지 않는다고 오류를 발생시킬 수 있기 때문이다.

 

3. kafaka 테스트

각 kafka폴더내 bin/window로 이동하여 아래 명령어들을 테스트할 수 있다.

 

• 토픽 생성
  

  kafka-topics.bat --create --zookeeper localhost:2181, localhost:2182, localhost:2183 --replication-factor 3 --partitions 1 --topic MyTopic​

• 토픽 리스트 확인
  

  kafka-topics.bat --list --zookeeper localhost:2181, localhost:2182, localhost:2183​

• 토픽 삭제
  

  kafka-topics.bat --delete --zookeeper localhost:2181, localhost:2182, localhost:2183 --topic MyTopic​

  
  
  • 토픽 삭제가 되지 않을 경우
    

    # zookeeper shell명령모드로 변경
    zookeeper-shell.bat localhost:2181
    
    # 토픽 확인
    ls /brokers/topics
    
    # 모든 토픽 삭제
    deleteall​

• 토픽 상세정보 확인
  

  kafka-topics.bat --describe --zookeeper localhost:2181, localhost:2182, localhost:2183 --topic MyTopic​

• 컨슈머 생성 (구독)
  

  kafka-console-consumer.bat --bootstrap-server localhost:9092, localhost:9093, localhost:9094 --topic MyTopic​

• 프로듀서 생성 (발행)
  

  kafka-console-producer.bat --broker-list localhost:9092, localhost:9093, localhost:9094 --topic MyTopic​

 

4. 마치며

이번에도 간단하게 작성되었으나 진행하는데 이상은 없을 것이라 생각한다. 기존에 kafka포트 방화벽 설정을 9092만 열어줬다면 9093, 9094도 열어두는 것을 잊지 말자, 그리고 kafka클러스터 내에서 토픽의 크기가 확장되는 방법이 파티션인데 파티션의 개수를 설정하는 것이 매우 중요하다, kafka클러스터 구성에서 kafka의 개수가 늘어날 때마다 클러스터 전체에 메시지가 균등하게 저장되도록 파티션의 개수를 설정하는 과정을 진행해야 한다. 한 파티션당 한 컨슈머가 메시지를 소비하기 때문에 컨슈머의 처리 속도와 목표 처리량에 따라 파티션 개수가 달라진다. 기본적으로는 클러스터 구성내 kafka의 개수와 같게 하거나 배수로 증가시키는 것이 좋다 얘를 들어 위의 작성내용과 같이 클러스터 내 kafka를 3대를 운영할 경우 파티션은 각 한 개씩 해서 총 3개로 설정을 하거나 각 2개 해서 6개 내지 3개씩 해서 9개.. 이렇게 구성하는 것이 좋다 초당 수백 건의 데이터를 처리할 수 있으니 운영하는 시스템에 따라 잘 확인해서 설정하는 것을 권장한다.