로컬 환경에서 Docker를 이용하여 Kafka 서버를 띄우고, Kafka Client를 통해 간단하게 메시지를 송수신하는 방법을 소개합니다.

카프카에 조금 더 익숙해지기 위해 로컬에 Kafka Server, Kafka Client 를 간단히 띄워보기로 했다. Kafka Server 는 Docker 를 이용하여 띄우고, Kafka Client (Producer, Consumer) 는 shell 파일을 다운로드 받아 로컬에서 간단히 실행해 볼 수 있다.

Kafka Server 설치

사실 로컬에 Kafka 를 다운로드 받아 로컬에 바로 띄우는게 가장 간단하지만, zookeeper 등의 설정을 단순화하기 위해 Docker 이미지를 통해 띄워보기로 했다. 이미 거의 비공식 표준으로 사용되고 있는 Kafka-Docker 레포지토리가 있다. 해당 레포지토리를 클론받아서 Docker 에 띄워주기만 하면 된다.

$ git clone https://github.com/wurstmeister/kafka-docker

docker-compose.yml 파일 수정

해당 레포지토리를 살펴보면, docker-compose 로 Kafka 설정이 되어있는 것을 확인할 수 있다. 왜냐하면, Kafka 는 Zookeeper 과 같이 띄워져야 하는데, 이러한 의존성들을 모두 설치/실행하기 위해서는 docker-compose 가 편리하기 때문이다.

또한, 로컬에서 테스트하는 환경이기 때문에, Kafka 분산 환경은 고려하지 않고 우선 한 대만 띄워보기로 하자.

Kafka-Docker 레포지토리에서, docker-compose-single-broker.yml 파일의 다음 설정을 수정한다.

-- docker-compose-single-broker.yml
...
KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
...

docker-compose 실행

$ docker-compose -f docker-compose-single-broker.yml up -d

docker 이미지가 제대로 띄워졌는지 확인

docker ps -a

Zookeeper, Kafka 두 이미지가 잘 띄워졌으면 정상적으로 설치된 것이다.

로컬에 Kafka Client 설치

이제 Docker 로 메세지를 받을 Kafka Server 는 잘 띄워졌다. 동작 확인을 위해 로컬에 Kafka 를 설치해서 메세지를 날려보기 위해서는 Kafka Client 도 설치가 필요하다. 위에서 클론받은 레포지토리의 Dockerfile 를 한 번 확인해보자.

ARG kafka_version=2.8.1
ARG scala_version=2.13

Kafka 와 Scala 버전을 확인했다면, 이에 맞게 Kafka Client (스크립트) 를 다음과 같이 설치한다. Kafka Download Page

$ wget https://archive.apache.org/dist/kafka/2.8.1/kafka_2.13-2.8.1.tgz
$ tar -xzvf kafka_2.13-2.8.1.tgz

Topic 생성

$ cd kafka_2.13-2.8.1
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test_topic

Producer 실행

$ bin/kafka-console-producer.sh --topic test_topic --broker-list localhost:9092

위와 같이 실행하면, Message 전송 대기 상태가 된다. 메세지를 전송하기 전에, Consumer 를 띄우자

Consumer 실행

Consumer 는 터미널 세션을 하나 새로 열어, 다음을 실행시킨다.

$ bin/kafka-console-consumer.sh --topic test_topic --bootstrap-server localhost:9092 --from-beginning

Producer 에서 메세지 전송 후 Consumer 에서 확인

Producer 터미널 세션에서 아무 텍스트나 타이핑을 해보면, Consumer 창에서 메세지를 수신한 것을 확인할 수 있다.

kafkajs 활용

위에서는 Kafka CLI Client 를 사용하여 테스트를 해 보았다. kafkajs 를 활용하여 Node.js 환경에서 테스트하는 방법을 살펴보자.

import { Kafka } from "kafkajs";

async function main() {
  const brokers = ["localhost:9092"];
  const groupId = "test_group";
  const topic = "test_topic";

  /** Create Kafka Instance */
  const kafka = new Kafka({
    brokers,
  });

  /** Consumer */
  const consumer = kafka.consumer({ groupId });
  await consumer.connect();
  await consumer.subscribe({ topic, fromBeginning: true });
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log({
        topic,
        partition,
        offset: message.offset,
        value: message.value?.toString(),
      });
    }
  });

  /** Producer */
  const producer = kafka.producer();
  await producer.connect();
  await producer.send({
    topic,
    messages: [{ value: "Hello World" }],
  });
}

main();

이것도 읽어보세요