kafka 시작하기

✏️ Kafka 와 Zookeeper 띄우기

📍 Compose 파일 생성하기

• kafka 와 Zookeeper 를 실행하는 compose 파일
  • docker-compose.yml 로 생성
    • project 디렉토리에 생성해주었다.
  • docker-compose up 로 컴포즈 파일 실행할 수 있다.
    • 파일에 volumes 을 선언했기 때문에 볼륨을 생성해줘야 한다.
    • 별도 경로를 입력하지 않았기 때문에 docker 가 관리하는 디렉토리에 볼륨을 생성 하면 된다.

version: '2'
services:
  zookeeper:
    image: 'bitnami/zookeeper:latest'
    container_name: zookeeper
    ports:
      - '2181:2181'
    volumes:
      - 'zookeeper_data:/bitnami'
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: 'bitnami/kafka:latest'
    container_name: kafka
    ports:
      - '9092:9092'
    volumes:
      - 'kafka_data:/bitnami'
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - ALLOW_PLAINTEXT_LISTENER=yes
    depends_on:
      - zookeeper
volumes:
  zookeeper_data:
    external: true
  kafka_data:
    external: true

• 제대로 설치가 되었고 실행까지 정상적으로 되었다.

CONTAINER ID   IMAGE                               COMMAND                  CREATED          STATUS          PORTS                                                NAMES
c32974adbf29   bitnami/kafka:3.3.2-debian-11-r38   "/opt/bitnami/script…"   32 seconds ago   Up 31 seconds   0.0.0.0:9092->9092/tcp                               kafka
5ac5ff5f986c   wurstmeister/zookeeper              "/bin/sh -c '/usr/sb…"   3 minutes ago    Up 3 minutes    22/tcp, 2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp   zookeeper

📍 Kafka 환경설정

• 카프카가 실행되는 터미널에 관리자 권한으로 접속한다.
  • bitnami 는 기본적으로 root 가 아니기 때문에 exec 명령어를 실행할 때 root 를 명시해 줘야 한다.

docker exec -it -u root kafka /bin/bash

• config 디렉토리로 의 server.properties 파일을 편집한다.

vim /opt/bitnami/kafka/config/server.properties

• 만약 vim 이 없다면 관리자 권한으로 다시 접속한다음 update 를 하고, vim 을 install 한다.

apt-get update
apt-get install vim

• vim 으로 server.properties 에 들어가서 Socket Server Settings 를 찾는다.
  • 아래처럼 생겼다.

############################# Socket Server Settings ############################# 
 
# The address the socket server listens on. If not configured, the host name will be equal to the value of 
# java.net.InetAddress.getCanonicalHostName(), with PLAINTEXT listener name, and port 9092. 
#   FORMAT: 
#     listeners = listener_name://host_name:port 
#   EXAMPLE: 
#     listeners = PLAINTEXT://your.host.name:9092 
#listeners=PLAINTEXT://:9092 
 
# Listener name, hostname and port the broker will advertise to clients. 
# If not set, it uses the value for "listeners". 
#advertised.listeners=PLAINTEXT://your.host.name:9092

• 해당 문서에서 listeners=PLAINTEXT://:9092 와,
  advertised.listeners=PLAINTEXT://your.host.name:9092 를 주석해제하고,
  your.host.name 에 공인 ip 를 입력한다.
  - 나는 로컬환경에서 테스트용으로 사용할거기 때문에 localhost 로 입력했다.
• 그 다음 delete.topic.enable=true 와, auto.create.topics.enable=true 를 추가해주고 저장한다.
• 그 다음 kafka 를 재시작 해주면 kafka 설정이 완료된다.

############################# Socket Server Settings ############################# 
 
# The address the socket server listens on. If not configured, the host name will be equal to the value of 
# java.net.InetAddress.getCanonicalHostName(), with PLAINTEXT listener name, and port 9092. 
#   FORMAT: 
#     listeners = listener_name://host_name:port 
#   EXAMPLE: 
#     listeners = PLAINTEXT://your.host.name:9092 
listeners=PLAINTEXT://:9092 
delete.topic.enable=true
auto.create.topics.enable=true
# Listener name, hostname and port the broker will advertise to clients. 
# If not set, it uses the value for "listeners". 
advertised.listeners=PLAINTEXT://localhost:9092

✏️ Kafka 연동 테스트

• 2개의 터미널로 각각 Consumer 와 Producer 의 역할을 맡길 수 있다.
  • kafka 를 docker 로 띄웠기 때문에 아래의 test 는 container 에 접속해서 실행해야 한다.

📍 Producer 연결

• 아래의 명령어를 입력해서 > 가 나오면 성공
  • exam-topic 이라는 토픽과 브로커를 생성해 프로듀서를 실행하는 명령어이다.

kafka-console-producer.sh --topic exam-topic --broker-list localhost:9092

📍 Consumer 연결

• 아래 명령어를 입력하고 아무것도 나오지 않으면 성공이다.
  • exam-topic 을 구독하는 컨슈머를 생성하는 명령어이다.

kafka-console-consumer.sh --topic exam-topic --bootstrap-server localhost:9092 --from-beginning

• 생성한 토픽은 아래 명령어로 확인할 수 있다.

kafka-topics.sh --list --bootstrap-server localhost:9092

• Procedure 에서 message 를 입력했을 대 Consumer 에 출력되면 성공이다.

📍 Kafka 의 브로커 리스트 확인

• 아 명령어로 현재 생성되어있는 브로커를 확인할 수 있다.

kafka-topics.sh --bootstrap-server localhost:9092 --list