[Kafka/Redis] Kafka 설치

1주차 - Kafka 개념
2주차 - Kafka 설치
3주차 - Kafka와 Spring
4주차 - Kafka 메트릭
5주차 - Redis 개념
6주차 - Kafka/Redis를 활용한 실시간 트랜잭션 집계 서비스 

 

 

1. Kafka 설치와 설정

1) Kafka 설치

서비스들의 호출 종속성이 카프카에만 생김 

 

- zookeeper 설치 

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

 

- broker 설치 (listener 포트 번호 변경) 

.\bin\windows\kafka-server-start.bat .\config\server.properties

 

- 토픽 생성 

.\bin\windows\kafka-topics.bat --create --topic test-topic --bootstrap-server localhost:9092
.\bin\windows\kafka-topics.bat --list --topic test-topic --bootstrap-server localhost:9092 (리스트확인

 

- 메세지 produce / consumer 

.\bin\windows\kafka-console-producer.bat --topic test-topic --bootstrap-server localhost:9092

엔터 누를 때마다 1개의 메세지 전송됨

.\bin\windows\kafka-console-consumer.bat --topic test-topic --bootstrap-server localhost:9092 --from-beginning

 

produce에서 메세지 발생하면 실시간으로 consumer에 노출됨

.\bin\windows\zookeeper-shell.bat localhost:2181 (주키퍼 쉘 접속)
-------------------------------------------------------------------------------------
ls /
[admin, brokers, cluster, config, consumers, controller, controller_epoch, feature, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper]
ls /brokers
[ids, seqid, topics]
ls /brokers/ids
[0]
ls /brokers/topics
[__consumer_offsets, test-topic] ✅ 프로듀스에서 오류가 생겼을 때 컨슈머가 참조하는 정보

 

 

2) Dokcer compose

- docker 설치 및 git clone

- docekr compose 설정 확인 

version: "3"

services:
  mysql_database:
    image: mysql:5.7
    environment:
      MYSQL_ROOT_PASSWORD: root
      MYSQL_DATABASE: wordpress_db
      MYSQL_USER: wordpress_user
      MYSQL_PASSWORD: changeme!
    volumes:
      - ./mysql:/var/lib/mysql

  wordpress:
    depends_on:
      - mysql_database
    image: wordpress:5.8.1
    ports:
      - "8080:80"
    environment:
      WORDPRESS_DB_HOST: mysql_database:3306
      WORDPRESS_DB_USER: wordpress_user
      WORDPRESS_DB_PASSWORD: changeme!
      WORDPRESS_DB_NAME: wordpress_db
    volumes:
       - ./wordpress_html:/var/www/html

docker compose up 하게 되면 로컬에 wordpress 인스턴스 뜨게 됨 

localhost로 접속

 

 

- broker cluster docker 파일 확인

version: '3'
services:
  zookeeper-1:
    hostname: zookeeper1
    image: confluentinc/cp-zookeeper:6.2.0
    environment:
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_CLIENT_PORT: 12181
      ZOOKEEPER_DATA_DIR: /zookeeper/data
      ZOOKEEPER_SERVERS: zookeeper1:22888:23888;zookeeper2:32888:33888;zookeeper3:42888:43888
    ports:
      - 12181:12181
      - 22888:22888
      - 23888:23888
    volumes:
      - ./zookeeper/data/1:/zookeeper/data ✅ 데이터가 로컬에 저장되도록 함 

	....

  kafka-1:
    image: confluentinc/cp-kafka:6.2.0
    hostname: kafka1
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:12181,zookeeper2:22181,zookeeper3:32181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:19092
      KAFKA_LOG_DIRS: /kafka
    ports:
      - 19092:19092
    volumes:
      - ./kafka/logs/1:/kafka

주키퍼와 카프카 인스턴스들이 설치됨 

한번 늘려놓은 파티션을 줄일 수가 없음 

 

 

3) Kafka 설정들여다보기 

https://kafka.apache.org/41/configuration/

############################# Server Basics #############################

# The role of this server. Setting this puts us in KRaft mode
process.roles=broker,controller

# The node id associated with this instance's roles
node.id=1

# List of controller endpoints used connect to the controller cluster
controller.quorum.bootstrap.servers=localhost:9093

############################# Socket Server Settings #############################

✅ listeners=PLAINTEXT://:9092,CONTROLLER://:9093

# The number of threads that the server uses for receiving requests from the network and sending responses to the network
✅ num.network.threads=3 -- 실제로 서버가 요청을 받고 내보내는

# The number of threads that the server uses for processing requests, which may include disk I/O
✅ num.io.threads=8 -- 서버가 요청 처리 디스트 i/o

# The send buffer (SO_SNDBUF) used by the socket server
✅ socket.send.buffer.bytes=102400


############################# Log Basics #############################

# A comma separated list of directories under which to store log files
✅ log.dirs=/tmp/kraft-combined-logs -- 브로터 로그 저장 

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
✅ num.partitions=1

############################# Log Flush Policy #############################

#   ✅  1. Durability: Unflushed data may be lost if you are not using replication.
#   ✅  2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#   ✅  3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.

# The number of messages to accept before forcing a flush of data to disk
✅ #log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
✅ #log.flush.interval.ms=1000 -- 1초마다 디스크에 플러시 

############################# Log Retention Policy #############################

# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168

 

팔로어가 응답을 안 보내면 ISR 에서 제거 - replica rag...

 

 

4) Kakfa 운영해 보기 (AKHQ)

도커에서 접속함

 

- produce 실행 (도커에 직접 접속해서 특정 포트 번호를 가진 카프카에 들어가서 수행함) 

docker exec -it fastcampus_kafka_handson-kafka-1-1 bash
kafka-console-producer --bootstrap-server localhost:19092 --topic fastcampus

콘솔에서 입력한 데이터 값이 들어옴

 

- consumer group 생성 

kafka-console-consumer --bootstrap-server localhost:19092 --topic fastcampus --group fastcampus_group

컨슈머 그룹 생성 및 프로듀서 데이터 실시간 모니터링

 

 

5) Kakfa 운영해 보기 (Kafka Burrow)

Consumer Lag : 프로듀서가 커밋한 오프셋과 컨슈머가 커밋한 오프셋의 차이 

컨슈머가 아직 처리하지 못한 프로듀서가 발행한 메세지의 개수

컨슈머 랙을 주기적으로 알람을 받고 모니터링 하기 위해 만들어진 툴 

 

Interval 

Sliding Window 

 

 

 

2. Kafka Data Pipeline 구축

1) 데이터 파이프라인 구축해보기

실시간으로 로그 수집 및 분석 : ELK 스택

ELK가 문제가 생겼을 때의 모니터링을 할 수 있는 파이프라인

 

• logstash   Kafka에서 로그를 가져와 → 가공해서 → Elasticsearch에 저장
  • Kafka에서 webapp-logs 토픽 구독
  • JSON 메시지 읽음
  • message 필드를 Apache 로그 패턴으로 분해
  • 구조화된 필드만 남김
  • Elasticsearch에 날짜별 인덱스로 저장

  • input { 👉 Kafka에서 데이터를 읽어오겠다
    	kafka {
    		bootstrap_servers => "kafka1:19092,kafka2:29092,kafka3:39092"
    		group_id => "logstash" ✅ Kafka consumer 그룹 이름
    		topics => ["webapp-logs"]
    		consumer_threads => 1
    		codec => json() ✅  Kafka 메시지를 JSON으로 파싱
    	}
    }
    
    ## Add your filters / logstash plugins configuration here
    filter {
    	grok { 👉 로그를 정규식 기반으로 구조화
    		match => {
    			"message" => "%{COMBINEDAPACHELOG}"
    			 ✅ 127.0.0.1 - - [10/Oct/2024:13:55:36 +0900] "GET /index.html HTTP/1.1" 200 2326 를 쪼갬 
    
    		}
    	}
    	remove_field => "message"
    }
    
    output {
    	elasticsearch {
    		hosts => "elasticsearch:9200"
    		user => "elastic"
    		index => "application-log-%{+YYYY.MM.dd}"
    		password => "changeme"
    		ecs_compatibility => disabled
    	}
    }

• pipeline yml 실행
  
  • apache server 설정 변경
  • filebeat -> kafka (output) 

 

2) Confluent Kafka 추가기능

Producer
   ↓
REST Proxy
   ↓
Kafka
   ↓
Schema Registry (스키마 검증)
   ↓
Consumer

• Rest Proxy

 

• Schema Registry
  • 문제가 되는 경우 (유효성 검증 없음)
  • Schema registry : 동일 스키마에 대한 호환성 체크를 위한 버전 유지
  • Kafka 메시지의 “설계도(스키마)”를 중앙에서 관리하는 서버  
• 호환성
  • 스키마를 바꿀 때, 기존 시스템을 보호하는 안전장치

  • # 처음 스키마 (v1)
    {
      "type": "record",
      "name": "account",
      "fields": [
        {"name": "balance", "type": "int"}
      ]
    }
    
    # 그런데 개발자가 이렇게 바꿈 (v2)
    {
      "type": "record",
      "name": "account",
      "fields": [
        {"name": "balance", "type": "int"},
        {"name": "currency", "type": "string"}
      ]
    }

  • 예) BACKWARD : 새로운 스키마가 이전 데이터를 읽을 수 있어야 함
    • Consumer가 새 버전이어도
    • Kafka에 저장된 옛날 메시지를 읽을 수 있어야 한다

 

 

3) Kafka Connect 활용

별도의 개발 없이 Kafka를 통해 Datasource/destination간 메세지 송수신을 가능하도록 해주는 솔루션 

 

docker로 서버를 띄운 후에  connector 등록 요청 

# mysql connector 등록
curl -v -XPOST -H'Accept:application/json' -H'Content-Type:application/json' http://connect1:18083/connectors \
  -d '
{
    "name": "mysql-source-connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "database.hostname": "${myip}",
        "database.port": "3306",
        "database.user": "root",
        "database.password": "passwd",
        "database.server.id": "1234",
        "database.server.name": "mysql-1",
        "database.include.list": "fastcampus",
        "database.history.kafka.bootstrap.servers": "${myip}:19092, ${myip}:29092, ${myip}:39092",
        "database.history.kafka.topic": "kafka-student-changes",
        "include.schema.changes": "true",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": "false",
        "value.converter.schemas.enable": "false"

    }
}'

# s3 connector 등록
curl -v -XPOST -H'Accept:application/json' -H'Content-Type:application/json' http://connect1:18083/connectors \
  -d '{
    "name": "s3-sink-connector",
    "config": {
      "topics": "mysql-1.fastcampus.kafka",
      "connector.class": "io.confluent.connect.s3.S3SinkConnector",
      "flush.size": 1,
      "s3.bucket.name": "fastcampus",
      "s3.region": "us-east-2",
      "s3.part.size": "5242880",
      "s3.proxy.url": "http://${myip}:4566",
      "format.class": "io.confluent.connect.s3.format.json.JsonFormat",
      "key.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "key.converter.schemas.enable": "false",
      "value.converter.schemas.enable": "false",
      "storage.class": "io.confluent.connect.s3.storage.S3Storage",
      "aws.access.key.id": "test",
      "aws.secret.access.key": "test",
      "topics.dir": "topicsdir"
    }
  }'

 

connector 관련 정보 curl 

+ Mysql 에서 update/insert 등의 쿼리를 날릴 때마다 -> S3에 output.txt 에 정보 담김 

 

 

3.  MSA 환경에서의 Kafka 활용

EDA (Evnet Driven Architecture)

ㄴ분산 시스템에서 비동기 통신 방식으로 이벤트를 발행/구독하는 아키텍처 

   ✔ 동기통신 : Restful API (peer to peer -> peer가 항상 살아있어야 함)

    ✔ 비동기 통신 : Message Broker, Kafka 통한 pub/sub

 

 

EDM 서비스 구현

EDM 에서 발생한 이벤트는 이벤트 스토어에 저장 (이벤트 로그)

비동기 통신을 사용하여 Mircro Service 사이의 느슨한 결합 가능

Service는 Produce / Consumer 코드 둘다 작성됨

 

1) transaction 처리

SAGA 패턴

 

2) 어플리케이션

• topic 생성 (파이선)

from confluent_kafka.admin import AdminClient, NewTopic

conf = {'bootstrap.servers': 'kafka1:19092'}
admin = AdminClient(conf=conf)
admin.create_topics([NewTopic('order', 1, 1), NewTopic('inventory', 1, 1), NewTopic('payment', 1, 1)])
print(admin.list_topics().topics)

• 주문 서비스: API

model = OrderModel(transaction_id, user_id)
model.updated = datetime.now()
model.status = 'ORDER_CREATED'
model.save()

• 주문 Consumer

consumer.subscribe([inbound_event_channel])
    while True:
        message = consumer.poll(timeout=3)
        if message is None:
            continue

        event = message.value()
        if event:
            logging.debug(event)
            evt = json.loads(event)
            status = evt['status']
            transaction_id = evt['transaction_id']
            inventory_id = int(evt['inventory']['id'])
            quantity = int(evt['inventory']['quantity'])

            if status == 'ORDER_CREATED':
                process_inventory_reserved(evt, transaction_id, inventory_id, quantity)
            elif status == 'ORDER_CANCEL':
                process_order_cancel(evt, transaction_id, inventory_id, quantity)
            else:
                logging.error('unknown event')

• 결제 Conumer

def process_consumer(is_pg_api_call_succeeded):
    conf_producer = {'bootstrap.servers': 'kafka1:19092'}
    producer = Producer(conf_producer)
    conf_consumer = {'bootstrap.servers': 'kafka1:19092', 'group.id': 'payment_consumer_group', 'auto.offset.reset': 'earliest'}
    consumer = Consumer(conf_consumer)

    inbound_event_channel = 'payment'
    outbound_success_event_channel = 'order'
    outbound_failure_event_channel = 'inventory'

    try:
        consumer.subscribe([inbound_event_channel])
        while True:
            message = consumer.poll(timeout=3)
            if message is None:
                continue

            event = message.value()
            if event:
                logging.debug(event)
                evt = json.loads(event)
                if is_pg_api_call_succeeded == 'True':
                    evt['status'] = 'PAYMENT_COMPLETE'
                    producer.produce(outbound_success_event_channel, value=json.dumps(evt))
                else:
                    evt['status'] = 'ORDER_CANCEL'
                    producer.produce(outbound_failure_event_channel, value=json.dumps(evt))
            else:
                continue