[Kafka]Java 17 + WebFlux + Docker 환경에서 카프카 실습 (1)

목차

 

한 달쯤 전에 올렸던 글에서, 야심 차게 MSK를 도입하겠노라고 선언을 했었다.

 

[Cloud]Java 17 + WebFlux 환경에 AWS MSK 도입하기(1) - MSK?

[Cloud]Java 17 + WebFlux 환경에 AWS MSK 도입하기(1) - MSK?

 

그 이후 이리 구르고 저리 구르다가 알게 된 사실은, 로컬 환경에서 MSK로 접근하는 건

 

불가능한 것은 아니지만 쉽지 않다는 것.

 

가장 쉽게 MSK에 접근하는 방법은 같은 VPC안에 EC2를 생성해서 접근하는 방법이라는 것..

 

그럼 실시간으로 디버깅하고 구현을 반영하는 건 어떻게 해? 그때그때 배포를 해야 하나?

 

하고 깨닫는데 2주 반 정도 걸린 것 같다.

 

그래서 일단은 로컬에서 충분히 테스트와 연습을 하고 그다음 MSK로 넘어가는 것으로 결정.

 

해서 오늘부터 카프카를 로컬 환경에서 돌리고, 점점 복잡한 로직을 구현하는 연습을 하려고 한다.

 

일단 최종 목표는 자바 프로젝트에서 메시지큐에 객체를 넣으면 해당 객체를 플라스크 프로젝트에서 받아

 

정해진 로직을 수행한 뒤 다시 객체로 돌려주는 것.

 

시작해 보자.

 

Docker+Kafka

 

카프카는 여타 소프트웨어와 비슷하게 도커로 설치 및 운영이 가능하고 또 편리하다.

 

물론 가장 최근의 기능을 사용하지는 못한다고 하지만, 역시 직접 설치하고 설정하는 건 귀찮으니까.

 

하지만 사실, 직접 받아서 설정하고 사용하는 것도 그다지 어렵지는 않았다.

 

따라서 이 부분은 개인 취향에 맞게.

 

최신 버전은 그렇지 않지만 구 버전에서는 카프카는 주키퍼를 사용해 여러 가지 작업을 수행한다.

 

따라서 도커로 설치할 때 주키퍼도 설치하고 실행하는 것이 필요하다.

 

이를 위해 docker-compose.yml 파일을 아래와 같이 적는다.

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

물론 여기선 도커와 도커 컴포즈가 설치되어 있다는 가정하에 진행한다.

 

위와 같이 작성 및 저장을 하고 해당 폴더에서 터미널을 열어

docker-compose up -d

를 실행하면

 

뭔가 잔뜩 다운로드하고 위와 같이 켜지는 것을 알 수 있다.

 

마지막 경고는 무시해도 된다.

 

계속해서 

docker ps

를 입력하면

 

위와 같이 카프카와 주키퍼가 제대로 실행되는 것을 확인할 수 있다.

 

여기서 만약 저 두 가지가 안 보인다면 뭔가 잘못된 것이니 docker-compose.yml 파일을 점검하거나

 

도커의 상태를 확인하도록 하자.

 

Topic

 

이어서 실습에 사용할 토픽을 만들어보자.

 

아래와 같은 명령어를 입력하면 실행 중인 도커로 진입할 수 있다.

docker container exec -it kafka bash

제대로 진입이 되었다면 다음과 같이 입력하자.

kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1

이렇게 하면 test-topic이라는 이름의 토픽이 생성된다.

 

만약 이름을 바꾸거나 해서 토픽을 삭제하고 싶다면 아래와 같이 입력한다.

kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test-topic

호스트 주소, 포트 주소, 토픽 이름 등을 잘 조정해야 하지만 이 글을 그대로 따라 했다면 이걸로 충분하다.

 

계속해서 테스트에 사용할 자바 프로젝트의 application.yml에 다음과 같이 추가한다.

spring:
  kafka:
    topic:
      myTopic: test-topic
    consumer:
      bootstrap-servers: localhost:9092
      group-id: test
      auto-offset-reset: earliest
    producer:
      bootstrap-servers: localhost:9092

여기서 그룹 아이디는 임의로 지정해도 되고,

 

earliest 설정은 큐에 들어온 순서대로, 그러니까 선입선출 방식으로 구독자가 받아간다는 뜻이다.

 

여기까지 하면 일단 기본 설정은 끝난다.

 

관련한 자바 설정과 구현은 다음 글에서 하기로 하자.

 

Append

 

추가로 컨테이너 내부에서 사용할 수 있는 명령어를 정리한다. 계속 추가될 예정이다.

 

존재하는 모든 토픽을 확인하는 명령어이다.

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

다음은 특정 토픽을 삭제하는 명령어이다.

kafka-topics.sh --bootstrap-server [Bootstrap Server 주소] --delete --topic [삭제할 토픽 이름]

이어서 특정 토픽에 들어있는 데이터를 로그처럼 실시간으로 볼 수 있는 명령어이다.

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