코루틴(Coroutine)

네이티브 코루틴(Native Coroutine)

파이썬에서 여러 작업을 동시에 병렬 처리하기 위해 비동기(Asynchronous) 함수를 사용한다.

async def cr_func():
    print('test')

이렇게 async 키워드로 만든 비동기 함수를 코루틴(Coroutine) 또는 코루틴 함수라고 한다. 그렇다면, 이 코루틴 함수는 언제 어떻게 사용하는 것일까?

async def cr_func():
    print('test')

cr_func()


### ----- Result ----- ###
RuntimeWarning: coroutine 'cr_func' was never awaited
### ------------------- ###

cr_func 이라는 이름의 코루틴 함수를 선언하고 호출하자 다음과 같은 에러가 나타났다. async 키워드로 선언된 코루틴 함수는 일반 함수 호출 방식과는 다르며, 다음과 같이 실행한다.

import asyncio

async def cr_func():  # 코루틴 함수 선언
    print('Coroutine Function!')

loop = asyncio.get_event_loop()       # asyncio 모듈의 event loop 객체 생성
loop.run_until_complete(cr_func())  # cr_func 함수가 완전히 종료될 때까지 대기
loop.close()                          # event loop 종료 


### ----- Result ----- ###
Coroutine Function!
### ------------------- ###

📌 asyncio 모듈의 get_event_loop를 얻어와서 run_until_complete의 인자로 코루틴 함수를 전달하여 실행한다. run_until_complete 함수는 인자로 받은 함수가 완전히 종료될 때까지 이벤트 루프를 실행하게 된다. 이와 같이, async로 선언된 코루틴을 "네이티브 코루틴" 이라고 한다.

다른 함수에서 이 코루틴 함수를 호출할 때는 어떻게 할까?

import asyncio

async def cr_func():
    print('test')

async def main():
    await cr_func()

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

코루틴 함수 생성 방식은 동일하나, 코루틴 함수를 호출하고자 하는 함수에서 await 키워드를 붙여줘야 한다. await = 말 그대로 cr_func()함수가 완료되기를 기다린 후 결과를 가져온다는 의미이다.

응답을 받을 때까지 아무것도 하지 않고 기다린다는 것이 아니라 해당 함수가 완료될 때까지 다른 작업을 하면서 완료를 기다린다는 의미이다.

제너레이터 기반 코루틴(Generator-based Coroutine)

일반적으로 사용하는 네이티브 코루틴 외의 또 다른 방식인 제너레이터 기반 코루틴에 대해 정리해보았다.

제너레이터에 대해서 정리한 글

다음은 yield 키워드를 통해 메인 루틴과 서브 루틴간에 값을 이동하며 실행의 지연 및 재개를 보여주는 예제이다.

def cr_gen():
    total = 0
    while True:    # 코루틴을 계속 유지하기 위해 무한 루프 사용
        num = yield total # total 발생 및 num에 값 저장
        total += num

cr = cr_gen()
print(next(cr)) # 처음 yield 까지 실행
print(cr.send(3)) # num에 3 저장 후 다음 yield 까지 실행
print(cr.send(2)) # num에 2 저장 후 다음 yield 까지 실행

""" 실행결과
0
3
5
"""

📌이와 같이 코루틴 바깥에서 보낸 값을 yield 변수(여기서는 num)에 받아 저장한 후, 다음 루틴을 실행한다. 정리하면, 핵심은 "코루틴은 yield에서 함수 중간에 대기한 다음에 메인 루틴을 실행하다가 다시 코루틴을 실행한다"는 것이다.

이터레이터(Iterator)

iterable한 객체를 내장 함수 또는 iterable객체의 메소드로 객체를 생성할 수 있다. 그렇다면 iterable한 객체란 무엇일까? iterable, 반복 가능한 객체이다. 순서대로 다음 값을 리턴할 수 있는 객체를 의미하며, 파이썬에서 대표적으로 iterable한 타입은 list, dict, set, str, bytes, tuple, range 등이 있다. 예를 들어, list 객체는 다음과 같이 반복문을 통해 값을 사용할 수 있다. 이렇게 내부 요소를 하나씩 리턴할 수 있는 요소를 iterable한 객체라고 할 수 있다.

a = [1, 2, 3]

for item in a:
    print(item)

iter함수를 통해 iterable 객체를 iterator로 만들 수 있으며, iterator로 변경 후에 __next__() 함수 호출이 가능하다.

# 1. iter() 함수 사용 전
print(a.__next__) # AttributeError 발생


# 2. iter() 함수 사용 후
a = iter(a)
#a = [1, 2, 3].__iter__()     # 같은 결과

print(a.__next__()) # 1 출력
print(a.__next__()) # 2 출력
print(a.__next__()) # 3 출력
print(a.__next__()) # StopIteration Exception

__next__() 함수를 호출해서 값을 반복적으로 꺼내 쓸 수 있고, 마지막 값까지 모두 꺼낸 후에는 StopIteration 예외가 발생한다.

제너레이터(Generator)

iterator를 생성해주는 함수이며, 함수안에 yield 키워드를 사용한다. yield가 호출되면 암시적으로 return이 호출되며, 한번 더 실행되면 실행되었던 'yield' 다음 코드가 실행된다. 다음의 예제를 확인해보자.

def simple_gen():
    yield "Hello"
    yield "World"

gen = simple_gen()
print(gen)             # Hello 출력
print(next(gen))    # World 출력
print(next(gen))    # StopIteration Exception

처음 gen을 호출할 때에는 simple_gen()함수를 실행하다가 yield 키워드를 만나면서 "Hello" 값을 return하고 해당 위치를 기억한다. 이후, next함수를 통해 다시 함수가 호출되면 앞서 return한 이후부터 코드를 실행하고 yield 키워드를 만나면 또 다시 "World" 값을 return한다.

이렇게 생성한 genertaor는 iterable한 객체가 되며 for문에서 사용할 수 있게된다.

def list_generator():
    a = [1, 2, 3]
    for i in a:
        yield i        #한번씩 값을 바깥으로 전달

gen = list_generator()
#case 1.
print(list(gen))    # [1, 2, 3] 출력
#------------------------------

#case 2.
print(next(gen))    # 1 출력
print(next(gen))    # 2 출력
print(next(gen))    # 3 출력
#------------------------------

이와 같이, 제너레이터는 함수를 끝내지 않은 상태에서 yield를 사용해 값을 바깥으로 전달할 수 있다. return은 반환 즉시 함수가 끝나지만 yield는 잠시 함수 바깥의 코드가 실행되도록 주도권을 양보하여 값을 가져가게 한 뒤 다시 제너레이터 안의 코드를 계속 실행하는 방식이다.

*모든 제너레이터는 이터레이터에 포함된다고 할 수 있다. *

1. 파이썬-카프카 라이브러리 설치

먼저, 파이썬에서 카프카를 사용하기 위해 라이브러리를 설치한다.

pip install kafka-python

2. producer 생성

토픽으로 데이터를 전송할 프로듀서를 생성한다. 여기서는 new-topic.txt 파일의 값을 보내는 방식으로 적용한다. (참고: 나는 카프카 테스트를 프로젝트 루트의 test 폴더에서 작업하였다.)

👀 브로커를 'localhost:9092'라고 세팅해주었으니, 당연히 로컬에서 zookeeper와 kafka를 실행한 상태에서 진행한다. 로컬에서 zookeeper와 kafka 실행하는 방법은 이전 게시글을 참고한다.

  # 전송할 데이터 파일
  # test/new-topic.txt

  -----start-----
  1
  2
  3
  ..
  ..
  9998
  9999
  10000
  -----end-----

# test/producer.py

from kafka import KafkaProducer
import json
import time
from csv import reader


class MessageProducer:
    broker = ""
    topic = ""
    producer = None

    def __init__(self, broker, topic):
        self.broker = broker
        self.topic = topic
        self.producer = KafkaProducer(bootstrap_servers=self.broker,
                                      value_serializer=lambda x: json.dumps(x).encode('utf-8'),
                                      acks=0,
                                      api_version=(2,5,0),
                                      retries=3
                                      )

    def send_message(self, msg):
        try:
            future = self.producer.send(self.topic, msg)
            self.producer.flush()   # 비우는 작업
            future.get(timeout=60)
            return {'status_code': 200, 'error': None}
        except Exception as e:
            print("error:::::",e)
            return e

# 브로커와 토픽명을 지정한다.
broker = 'localhost:9092'
topic = 'new-topic'
message_producer = MessageProducer(broker, topic)

with open('test/'+topic+'.txt', 'r', encoding='utf-8') as file:
    for data in file:
        print("send-data: ", data)
        res = message_producer.send_message(data)

3. consumer 생성, 가져온 데이터 처리(테이블 insert)

토픽으로부터 데이터를 가져올 컨슈머를 생성하고, 컨슈머에서 데이터 처리를 한다. 수집한 데이터는 TblKafkaTest 테이블에 row 단위로 insert 한다. 테이블은 임시로 다음과 같이 생성하였다.

# test/models.py
from django.db import models

class TblKafkaTest(models.Model):
    idx = models.BigAutoField(primary_key=True)
    name = models.CharField(max_length=50, blank=True, null=True)
    col_a = models.CharField(max_length=50, blank=True, null=True)
    col_b = models.CharField(max_length=50, blank=True, null=True)
    col_c = models.CharField(max_length=50, blank=True, null=True)
    col_d = models.CharField(max_length=50, blank=True, null=True)
    create_date = models.DateTimeField()

    class Meta:
        managed = False
        db_table = "tbl_kafka_test"

# test/consumer.py

from kafka import KafkaConsumer
import time
import json
import datetime
from models import *

class MessageConsumer:
    broker = ""
    topic = ""
    group_id = ""
    logger = None

    def __init__(self, broker, topic, group_id):
        self.broker = broker
        self.topic = topic
        self.group_id = group_id

    def activate_listener(self):
        # 처리 시간 등의 결과 확인을 위해 kafka_output.txt 파일에 값을 이어 출력한다.
        sys.stdout = open('kafka_output.txt','a', encoding='utf-8')
        consumer = KafkaConsumer(
            bootstrap_servers=self.broker,
            group_id=self.group_id,
            consumer_timeout_ms=2000,
            auto_offset_reset='latest',
            enable_auto_commit=False,
            value_deserializer=lambda m: json.loads(m.decode('ascii'))
        )
        consumer.subscribe(self.topic)
        tot_start = time.time()
        start = time.time()
        i = 0

        try:
            with open(self.topic+'.txt', 'a', encoding='utf-8') as file:
                for message in consumer:
                    message = message.value
                    file.write(str(message))
                    i = i + 1

                    data = {
                        'name': topic,
                        'col_a': i,
                        'col_b': message,
                        'create_date': datetime.datetime.now()
                    }
                    TblKafkaTest.objects.create(**data)

                    consumer.commit()

                tot_end = time.time()
                tot_elapsed = tot_end - tot_start
                per_time_value = i / tot_elapsed

                print("=====================================================")
                print("총 처리 시간: ", tot_elapsed)
                print("총 처리 건수: ", i)
                print("초당 처리 건수: ", per_time_value)
        except KeyboardInterrupt:
            print("Aborted by user...")


broker = 'localhost:9092'
topic = 'new-topic'
group_id = 'consumer-1'

consumer1 = MessageConsumer(broker, topic, group_id)
consumer1.activate_listener()

consumer2 = MessageConsumer(broker, topic, group_id)
consumer2.activate_listener()

4. 실행 및 결과 확인

파이썬 인터프리터에서 producer.py 파일을 실행하였다. 전송 데이터 확인을 위해 print문을 넣었고, 터미널에 다음과 같이 출력되었다. new-topic.txt 파일에 1~10,000까지의 값이 순차적으로 모두 전송되었음을 확인할 수 있다.

데이터 전송은 완료되었으니, 컨슈머에서 확인을 해보자. 이번에는 consumer.py 를 실행하였고, kafka_output.txt 에서 출력 결과를 확인하였다.

# test/kafka_output.txt
=====================================================
총 처리 시간:  10.073350191116333
총 처리 건수:  20003
초당 처리 건수:  1985.7345987674096
=====================================================
총 처리 시간:  9.726698160171509
총 처리 건수:  0
초당 처리 건수:  0.0

✍ 나는 컨슈머를 2개로 설정해서 2번의 출력 결과를 얻었다. 그리고 consumer1 이 문제 없이 데이터를 모두 수집하였기 때문에, consumer2가 이어 받아 할 작업은 따로 없어 consumer2의 처리 건수는 0이 된다.

결과를 확인해보니 총 약 20,000 건의 데이터가 10초동안 처리되었다. 평균적으로 초당 1,985개의 데이터가 처리된 것이다. 엄청난 속도인 것 같다. 😱

테이블에 insert도 잘 되었을까? 🤔 역시 20,003개의 row 모두 잘 저장되었음을 확인할 수 있다.

추가적으로

나는 컨슈머 그룹을 설정하고 동일한 그룹ID로 consumer1 , consumer2를 생성하였다. 이렇게 하면 consumer1이 작업을 진행하다가 중단되어도 consumer2가 이어 받아 작업할 수 있게 된다. 단, 여기서 컨슈머 옵션은 auto_offset_reset='latest' 이어야 한다. 기타 컨슈머의 옵션은 잘 정리된 링크가 있어 첨부한다. 컨슈머 옵션 관련 링크

데이터 가져와 테이블에 저장하기 실습으로 간단하게 확인해보았지만, 실제로 수만 개의 데이터 처리를 할 때에는 매우 효과적이고 빠를 것 같다는 생각이 들었다.

stateful(상태 유지)

통신에 필요한 데이터를 서버가 가지고 있으며, 세션 종료시까지 클라이언트의 상태 정보를 저장하는 것을 의미한다. 작업중이던 서버 중단 시, 중단된 시점부터 다시 진행이 가능하다. 하지만, 서버 1과 통신 중에 해당 서버에 장애가 생겨 서버 2가 이어 받을 경우, 서버 2는 상태 정보를 가지고 있지 않기 때문에 통신에 문제가 된다. (재인증 절차가 필요하다.) 또한, 서버 확장(scale out) 작업 시에 기존의 데이터와 세션을 모두 옮겨주는 등의 관리가 필요할 수 있다.

stateless(무상태)

서버가 클라이언트의 상태 정보를 저장하지 않는다. (요청에 대한 응답만 처리) 통신에 필요한 데이터는 클라이언트가 가지고 있다가 서버와 통신 시 데이터를 전송하는 방식이다. 서버는 별도의 상태를 저장하지 않고, 해당 요청에 단순 응답만 한다. 그렇기 때문에 서버 1에 장애가 생기더라도, 서버 2가 이어 받아 통신을 진행하는 데에 문제가 없다. 상태 정보는 클라이언트가 가지고 있기 때문이다. 서버 확장 시에도 수월하게 대처 가능하다는 장점이 있으며, 클라이언트 요청에 많은 데이터가 소모될 수 있다는 점도 고려해야 한다.

📌인프라(Infra)* 란? IT 환경을 운영하고 관리하는 데 필요한 소프트웨어, 하드웨어, 서비스 및 IT 자원의 조합을 의미한다.

서버를 운영, 관리하다 보면 이용자 수의 증가, 부하 증가 등의 이유로 고성능의 서버가 필요하게 된다. 이러한 문제를 다음 2가지 방안으로 해결할 수 있다.

스케일 업(Scale-Up)

스케일 업은 말 그대로 기존 서버의 사양을 향상시키는 것이다. 기존 서버에 CPU나 RAM 등을 추가하거나 고성능의 부품, 서버로 교환하는 방식이다. "수직 확장"이라고도 한다. 성능 확장에는 한계가 있고, 성능 확장에 따른 비용 증가 폭이 큰 편이다. 또한, 한 대의 서버에 모든 부하가 집중되므로, 서버에 장애 발생시 영향도가 크게 작용한다. 하나의 서버에서 모든 데이터가 관리되는 데이터베이스 서버에 적합한 방식이다.

스케일 아웃(Scale-Out)

스케일 아웃은 서버 대수를 늘려 성능을 확장하는 방식이다. "수평 확장" 이라고도 하며, 서버의 부하를 여러 대가 균등하게 나누어 갖게 된다. 또한, 서버 한 대가 장애로 다운되더라도 다른 서버로 서비스 제공이 가능하다. 서버 대수 증가 비용 부담은 비교적 적으며, 분산 처리에 적합하다. 모든 서버가 동일한 데이터를 갖고 있어야 하므로, 데이터의 변화가 빈번하지 않은 웹 서버에 적합하다.

애자일 (Agile) "반복 접근 방식"

짧고 점진적인 주기로 개발하는 방법론이며, 제품이나 서비스 개발을 지속적으로 향상시킨다. 일정한 주기를 바탕으로 끊임없이 프로토타입을 만들어 고객의 요구사항을 만족시키는 방식이다.

(요구사항 분석 -> 설계 -> 개발 -> 테스트) * N -> 배포 작은 기능 단위로 짧은 주기로 빠르게 단위 테스트를 거친 후 보완하는 방식을 여러 번 걸친다.

• 각각의 한 주기 = 스프린트(Sprint)

장점

• 고객 중심적이며, 환경 변화에 잘 적응한다.
• 진행 속도가 빠르며, 결과물의 만족도가 높아진다.
• 잛고 반복적인 스프린트 과정을 거치기 때문에 개발 과정 중에 서비스 변경이 자유롭다.

단점

• 강력한 팀워크와 팀원 모두에게 책임감이 요구된다.
• 빠른 반복작업에 익숙한 스크럼 마스터가 필요하다.

워터폴 (Waterfall) "순차 접근 방식"

대부분의 회사에서 많이 사용하는 방법론으로, 프로젝트 시작부터 마감까지 일련의 순서에 따라 이뤄지는 방식이다. 프로세스 주기나 순서가 거의 일정하다. 업무 프로세스가 잘 정착된 환경에서 업무를 능숙하게 처리한다.

요구사항 분석 -> 설계 -> 개발 -> 테스트 -> 배포 전체 프로젝트를 하나의 과정으로 순차적으로 진행된다.

장점

• 프로세스가 길고 순서가 정해져있어 팀 규모와 상관없이 안정적으로 진행할 수 있다.
• 초기 단계에 요구사항과 예산이 정해져있어, 기술적인 리스크를 최소화한다.

단점

• 예측하지 못한 리스크에 많은 비용이 발생한다.
• 요구사항이 초기에 정해져 변경에 자유롭지 못하다.
• 순차적인 진행 방식에 따라, 이전의 과정이 완료되어야 다음 과정으로 진행이 가능해 개발 속도가 느리다.

📌 결론

두 유형 모두 각각의 장단점이 있으며, 방법론에 있어 정답은 없다. 프로젝트의 목표 및 일정 등에 따라 각 조직에 따라 프로젝트 개발에 적합한 방법론을 선택해 진행하면 된다.

마치며..

나는 지금까지 어떠한 방법론을 정하고 작업하진 않았지만, 두 방법론을 배워가면서 정리해보니 워터폴의 방식으로 프로젝트를 진행해왔구나 라는 생각이 든다. 아마도 대부분의 일반적인 기업에서는 워터폴의 방식으로 진행하지 않을까 싶다. 여러 스프린트를 거쳐 고객의 요구사항을 만족시킬 수 있는 애자일의 방식을 도입하는 것, 어렵지만 충분히 고려해볼 문제인 것 같다.😎

카프카의 기본 구성 요소 등을 모두 학습했다는 가정하에 작성하였다.

실행 환경: Window 기본 설치 요소: Java, Zookeeper, Kafka kafka 설치 폴더 경로: C:\kafka\kafka_2.13-3.3.1

📌 실행에 필요한 기본 환경 구축

👉 Java 설치

주키퍼는 독립적으로 사용이 불가능하며 자바에 의존성이 있기 때문에, 오라클 자바를 설치해야 한다. 설치 후 경로를 복사하여 시스템 속성 > 고급 > 환경 변수 > 시스템 변수 항목에 다음을 추가해준다.

 JAVA_HOME = C:\Program Files\ojdkbuild\java-1.8.0-openjdk-1.8.0.332-1 (본인의 자바 설치 경로)

👉 Zookeeper 설치

• https://archive.apache.org/dist/zookeeper/stable/
• 카프카를 띄우기 위해 반드시 실행해야 하는 주키퍼를 다운로드한다.

👉 Kafka 설치

• https://kafka.apache.org
• 위의 경로에서 tgz 파일을 다운받은 후 적절한 경로에 압축을 해제한다.

📌 실행해보기

1. Zookeeper 서버 띄우기

• 윈도우 cmd 창을 열고 주키퍼를 띄운다.
• kafka 설치 경로에서 실행한다. (예제의 경우, C:\kafka\kafka_2.13-3.3.1)

  C:\kafka\kafka_2.13-3.3.1> .\bin\windows\zookeeper-server-start.bat config\zookeeper.properties

  

그러면 위와 같이 이런 저런 메시지들이 나오고, 주키퍼가 정상 실행 되었음을 알리는 zookeeper 문구를 확인할 수 있다.그리고 하단으로 쭉~스크롤하면 커서가 깜빡깜빡 하고 있을 것이다. 여기까지 왔다면 주키퍼 띄우기 성공이다.

2. Kafka 서버 띄우기

• 위에서 띄운 cmd 창은 그대로 두고, 다른 cmd 창을 열어 이번에는 kafka를 띄워본다.
• 역시 이전과 동일한 경로에서 진행한다.

  C:\kafka\kafka_2.13-3.3.1> .\bin\windows\kafka-server-start.bat config\server.properties

  

❌ 그런데 나는 이 과정에서 에러를 마주했다. ❌1에서 실행한 주키퍼 서버에서 "java.io.IOException: 현재 연결은 원격 호스트에 의해 강제로 끊겼습니다" 와 같은 에러 메시지가 나타났고, 카프카 서버 실행을 시도했던 cmd 창에서도 다음과 같은 메시지와 함께 이 과정에 실패하였다.

구글링 후, 다음과 같은 방법으로 해결하였다.

..
log.dirs=/tmp/kafka-logs
..

카프카 설치 디렉토리/config/server.properties 파일의 log.dirs 항목에서 설정되어 있는 카프카 log path로 가보면 meta.properties라는 파일이 있다. 해당 파일을 지워주고 카프카를 재시작하였다.

Created topic new-topic.

C:\kafka\kafka_2.13-3.3.1\bin\windows> .\kafka-console-producer.bat --broker-list localhost:9092 --topic new-topic

C:\kafka\kafka_2.13-3.3.1\bin\windows> .\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic new-topic --from-beginning

Offset (오프셋)

오프셋이란, 각 파티션마다 메시지가 저장되는 위치를 의미한다. 당연히 오프셋 값은 '파티션 내에서' 고유하고, 순차적으로 표기된다. 테이블의 pk (id) 개념과 비슷하다고 볼 수 있다.

컨슈머가 메세지를 어디까지 읽었는지 저장하고 있기 때문에 다음에 읽을 메시지 정보를 쉽게 알 수 있다.

예를 들어, 현재 offset = 3이라면, 컨슈머가 0~2까지 메시지를 읽은 것이고 다음에 읽어야 하는 메세지 정보는 3번 이라는 것을 의미한다. 이러한 offset을 통해 데이터의 순서가 보장되는 것이다.

Consumer Group (컨슈머 그룹)

말 그대로 컨슈머들의 집합, 여러 컨슈머들이 하나로 묶인 논리적인 그룹이다. 한 컨슈머 그룹 내에서 1번 컨슈머에서 장애가 발생할 경우, 같은 그룹 내의 2번 컨슈머가 이어 받아 데이터를 계속해서 받아올 수 있다.

Rebalancing (리밸런싱)

리밸런싱 관련 검색을 하던 도중 이해하기 쉬운 이미지가 있어 가져왔다. (원본 링크) 리밸런싱이란, 위의 컨슈머 그룹에서 설명한 예와 같이 한 컨슈머로부터 다른 컨슈머가 소유관을 이관받는 작업을 의미한다.

컨슈머 그룹 내의 컨슈머들은 자신들의 파티션 소유권을 서로 공유한다. 이러한 리밸런싱은 컨슈머 그룹의 가용성과 확장성을 높여준다.

리밸런싱이 발생하면 컨슈머들은 메시지를 읽을 수 없어 해당 컨슈머 그룹 전체가 사용 불가 상태가 되어버린다. 리밸런싱이 발생하지 않는 것이 가장 좋지만,

❓ 그렇다면, 리밸런싱은 언제 발생할까?

컨슈머 그룹이 생성되면 컨슈머 그룹별로 관리하는 브로커가 지정되는데, 백그라운드 프로세스로 실행되는 이 브로커를 컨슈머 그룹 코디네이터(Consumer Group Coordinator) 라고 한다. 이 코디네이터가 컨슈머 그룹을 관리하면서 다음과 같은 상황이 발생되면 리밸런싱이 실행된다.

1) session.timeout.ms 설정시간에 heartbeat 시그널을 받지 못하는 경우

2) max.poll.interval.ms 설정시간에 poll() 메소드가 호출되지 않는 경우

• 컨슈머는 메시지를 가져오기 위해 브로커에 poll()요청을 보내고, 컨슈머는 가져온 메시지를 처리한 후 해당 파티션의 offset을 커밋한다. 이미지 출처:사람인 기술 블로그 poll()요청을 보낸 후 다음 요청을 다시 보내기까지의 시간이 [max.poll.interval.ms]의 기본값인 300000(5분) 보다 늦으면 브로커는 컨슈머에 문제가 있다고 판단하여 리밸런싱을 발생시킨다.

이번에 대용량 데이터를 실시간으로 처리하는 업무를 맡았다. 개발 과정에서는 특별한 무리는 없었지만, 실제 서비스가 될 경우 유저가 많아질수록 트래픽이 많이 발생하는 현상이 우려되었다. 이를 효과적으로 처리하기 위해 도입한 것이 바로 카프카(Kafka)이다.

"Kafka"란 무엇일까?

2011년 미국 링크드인(Linkedin)에서 개발한 것이다. 데이터 파이프라인, 분산 스트리밍, 실시간 스트리밍 데이터를 효과적으로 처리하기 위한 고성능 분산형 게시-구독 메시지 플랫폼이다.

카프카(Kafka)는 Pub-Sub 모델의 메시지 큐 형태로 동작하며 분산 환경에 특화되어 있다.

Pub-Sub 모델이란? 일반적으로 메시징 서비스 모델은 Queue 방식과 Pub-Sub 구조로 구분되는데, 카프카는 후자에 해당한다. Queue는 송신자와 수신자가 1:1 관계이다. 단일 송신자가 단일 수신자에게 데이터를 전송한다. Pub-Sub 구조는 송신자와 수신자가 m:n 관계이다. 즉, 여러 송신자가 메시지를 발행하고, 여러 수신자가 메시지를 구독한다.

한 마디로, 실시간으로 많은 데이터를 처리해야 할 때 빠르고 효과적인 분산 시스템이라고 할 수 있다.

현재 LINE, 링크드인에서 카프카를 도입해 사용하고 있으며, 국내에서도 사용량이 증가하는 추세이다.

kafka의 구성 요소

대략 어떤 시스템인지는 이해가 되었으니, 구성 요소를 먼저 알아보자. 이미지 출처: 원본 링크 이해하기 쉽게 표현된 이미지가 있어 해당 링크에서 가져왔다.

• Producer (프로듀서)

  • 메시지(이벤트)를 생성해 카프카 클러스터에 전송한다.

• Topic (토픽)

  • 메시지를 논리적으로 묶은 단위이다. (예를 들어 데이터베이스의 테이블, 파일에서 폴더에 해당)
  • 프로듀서가 메시지를 보낼 경우 저장되는 장소이다.
  • 하나의 토픽은 여러 개의 파티션으로 이루어져 있다.

• Partition (파티션)

  • 토픽 내에 분리된 공간이며, 각 토픽 당 데이터를 분산 처리하는 가장 작은 단위이다.
  • 조금 더 쉬운 이해를 위해 예를 들자면 토픽을 톨게이트, 파티션을 n차선이라고 비유한다.

• Broker (브로커)

  • 각각의 카프카 서버, 여러 개의 브로커 생성이 가능하다.
  • 메시지를 저장하고 관리한다.

• Consumer (컨슈머)

  • 메시지를 구독해 데이터를 받아온다.
  • 하나의 컨슈머에서 여러 개의 토픽 데이터를 받아올 경우, 처리하는 데 버거울 수 있다. 이럴 경우 같은 컨슈머 그룹 내에 또 다른 컨슈머를 추가해 작업을 각각 수행하면 더욱 효율적인 결과를 낼 수 있다.

• Zookeeper (주키퍼)

  • 분산 메세지 큐의 메타 데이터 정보를 관리한다. (브로커id, 컨트롤러id 등)
  • kafka를 띄우기 위해서는 반드시 실행되어야 한다.
  • Zookeeper는 홀수의 서버로 작동하도록 설계되어 있다. (최소 3, 권장 5)

kafka의 특징

• 고성능 : 프로듀서와 컨슈머가 브로커를 통해 데이터를 주고 받을 때, 한 번에 대용량의 데이터를 전송 가능하다. 또한, 컨슈머의 개수를 늘리면서 병렬 처리까지 가능하니 효율성이 높아진다고 볼 수 있다.

• 고가용성 및 확장성 : 카프카는 클러스터(*각기 다른 서버들을 하나로 묶어서 하나의 시스템같이 동작하게 함)로 동작 함으로로써 클라이언트들에게 고가용성의 서비스를 제공한다. 또한, 서버를 수평적으로 늘려 안정성과 성능을 향상시키는 스케일 아웃(Scale-out)이 가능하다. 브로커 서버를 여러 개로 운영하기 때문에, 일부 서버에 문제가 발생해도 나머지 서버에 영향 없어 중단되지 않고 동작 가능하다.

• 디스크에 저장 : 카프카는 메시지를 순차적으로 디스크에 저장한다. 그렇기 때문에, 서버에서 장애가 나더라도 실제 메시지는 디스크에 존재하기 때문에 데이터 유실 걱정이 없다.

• 분산 처리에 특화 : 여러 개의 파티션을 서버에 분산시켜 나누어 처리하기 때문에 속도가 향상된다.

📌 알아 둘 사항

• 한 토픽의 파티션 개수보다 더 많은 수의 컨슈머를 추가하는 건 의미가 없다.
• 파티션이 많다고 무조건 좋은 것은 아니다.
• 하나의 파티션을 두개 이상의 컨슈머가 소비할 수 없다.

이렇게 정리하고 보니, 대용량 및 실시간 처리에 적합한 카프카가 사용될 수 있는 곳이 많을 것 같다. 이런 훌륭한 기능을 가진 시스템을 사용하지 않을 이유가 없다. 실시간으로 전송되는 알림, 접속 로그, 실시간 집계 등의 처리에서 아주 빠르고 효과적으로 개선될 수 있을 것 같다. 카프카의 오프셋, 리밸런싱, 다중 컨슈머 그룹 등에 대한 내용은 추가적으로 다루도록 한다.😄

추가적으로...

카프카의 메타데이터 관리 도구인 주키퍼(Zookeeper)가 점차 제거된다고 한다. 관련 링크

🤚 이전 글에서 다뤘던 주요 내용에 대해 다시 정리해보자. 파이썬 어플리케이션과 서버를 연결해주는 "WSGI" 동기 함수만을 처리해 여러 작업 동시 수행에는 한계가 있었다. 이르 보완하기 위한 것이 바로 "ASGI"였다. ASGI는 기본적으로 요청을 비동기로 처리한다. 이 ASGI의 대표적인 예가 uvicorn이었다.

Django Ninja의 Async

본론으로 들어가서, Django Ninja의 Async 예제를 살펴보자. 기본적으로 장고의 ORM은 sync(동기적)로 작동한다. 우리는 async로 작동하도록 구현하기 위해 기존 방법과는 다르게 아래와 같이 2가지 방법을 통해 Sync(동기) -> Async(비동기) 로 설정 가능하다.

아래 2가지 모두 동일하게 asgiref 에서 제공하는 sync_to_async 모듈을 import 해준다.

☝ 함수로 사용

from asgiref.sync import sync_to_async 

async_function = sync_to_async(sync_function, thread_sensitive=False)
async_function = sync_to_async(sensitive_sync_function, thread_sensitive=True)

동기화 함수는 모두 메인 스레드에서 실행된다고 가정하며, 2가지 스레드 모드가 있다.

여기서 thread_sensitive 라는 값이 있는데, 기본 값은 True이다. True로 설정하면 다른 기능들과 동일한 기존 스레드에서 실행되고, False로 설정하면 완전히 새로운 스레드에서 실행되며 호출이 완료되면 닫힌다.

✌ 데코레이터로 사용

from asgiref.sync import sync_to_async

@sync_to_async
def sync_function(...):
   ...

이렇게 sync 함수 상단에 데코레이터로 선언해주고 sync 함수 그대로 작성하면 해당 함수가 async로 작동하기 때문에 더욱 간편하게 사용 가능하다.

Django-ninja에서 사용하기

Django-ninja에서는 router를 이용해 URL을 설정한다. 이 때, sync_to_async 데코레이터와 함께 사용하려면 다음과 같은 순서에 맞춰 적용한다.

sync_to_async 데코레이터를 먼저, path를 나중에 작성!

@sync_to_async
@router.get("/test", response=TestSchema)
def test_func(request, input):
    ...

참고

https://docs.djangoproject.com/en/4.1/topics/async/

Django Framework 기반의 백엔드 어플리케이션을 위한 WSGI 서버이다.

그럼 WSGI는 뭘까? 👉WSGI (Web Server Gateway Interface): Python 웹 어플리케이션이 웹 서버와 통신을 하기 위한 인터페이스이다. 웹 서버의 요청을 받아 Python 어플리케이션에 전달하는 역할을 한다.

이러한 WSGI의 대표적인 예 중 하나가 Gunicorn이다. 이 외에도, uWSGI, mode_wsgi, CherryPy 등이 있으며, 아래 이미지와 같이 웹 서버와 어플리케이션 서버 사이의 요청을 처리하는 역할을 한다. 이미지 출처: https://leffept.tistory.com/345

이렇게 파이썬 어플리케이션의 실행을 도와주는 WSGI에도 단점이 있다. WSGI는 요청을 받고 응답을 반환하는 동작이 단일 동기 호출 방식으로 처리 된다는 점이다. 오랜 시간 연결을 유지해야 하는 Websocket이나 긴 HTTP 요청을 처리하기에 이 방식이 적합하지 않다.

이와 같은 단점을 보완하고자 등장한 것이 ASGI(Asynchronous Server Gateway Interface) 이다! WSGI보다 나중에 나온 인터페이스로 WSGI의 단점을 보완하고, 비동기 코드를 처리하는 데에 효과적이다. 클라이언트로부터 여러 이벤트를 주고 받을 수 있고, 백그라운드 코루틴을 실행할 수 있다.

이 쯤에서, Uvicorn을 알아보자!

앞서 설명한 비동기 인터페이스를 통해 구현된 ASGI 웹 서버가 uvicorn이다. 단일 프로세스에서 uvloop 기반 비동기 Python code를 실행한다. 하지만, 단일 프로세스이기 때문에 서버 성능 저하가 발생할 수 있다. 그래서 우리는 웹 서버이자 프로세스 관리자인 gunocorn을 활용해 여러 개의 uvicorn을 관리하여 멀티 프로세스 환경을 구성한다.

아래의 구동 예시를 살펴보자.

gunicorn main:app --workers 3 --worker-class uvicorn.workers.UvicornWorker --daemon --access-logfile ./log.log

--workers : 프로세스 개수 (최대 vcpu 개수 * 2만큼 설정을 권장) --worker-class : 프로세스를 다중으로 실행하기 위해 필요한 옵션 --daemon : 백그라운드로 실행 --access-logfile ./access_log.log : access_log.log 이름으로 로그를 저장

구동 방법은 알겠다. 그런데 왜 멀티 프로세스로 돌리는거지? 위에 작성한 내용과 관련이 있다. 하나의 요청을 받고, 해당 요청을 수행하는 과정이 오래 걸리면, 그 시간 동안 다른 요청을 처리하지 못하게 된다. 실제 서비스하는 환경에서 이러한 이슈가 발생한다면...🤦‍♀️ worker의 개수를 늘려 여러 개로 병렬 처리를 한다면 위와 같은 문제는 발생하지 않을 것이다.

먼저, ORM이란 무엇일까?

Object Relational Mapping, 객체-관계 매핑을 뜻하는 용어로, 객체와 관계형 데이터베이스의 데이터를 자동으로 연결해주는 것을 의미한다.

👉 select_related : 정참조

Foreign Key 속성이 있는 객체에서 해당 부모를 참조하거나 1:1 관계에서 참조하는 경우

Query문의 Join을 사용해 foreign-key(one to one, many to one)를 참조하며, 관련된 테이블 정보를 받아올 수 있다.

다음의 예시와 함께 확인해보자. "나는 user id가 1인 사람의 도시 명을 알고 싶다." (User: 자식 테이블, City: 자식이 참조하는 부모 테이블)

# 도시 정보
class City(models.Model):
    id = models.BigAutoField(primary_key=True)
    name = models.CharField(max_length=20)

# 유저 정보
class User(models.Model):
    id = models.BigAutoField(primary_key=True)
    city = models.ForeignKey('City', on_delete=models.CASCADE)
    name = models.CharField(max_length=20)
    age = models.SmallIntegerField()

1. select_related 사용 X

  city_info = User.objects.get(id=1).city
  city_name = city_info.name

2. select_related 사용 O

  city_info = User.objects.select_related('city').get(id=1).city
  city_name = city_info.name

위의 1과 2 방법은 쿼리 상으로는 다를 게 없어 보인다. 오히려 2번이 더 복잡해 보이기도 한다. 그렇다면, DB 접근 방식을 비교해보자.

1번의 경우, DB에 2번의 접근을 하게 된다. User 테이블에서 id가 1번인 유저의 city 값을 가져오기 위한 조회 -- (1), 가져온 1의 데이터를 바탕으로 City 테이블을 조회해 name 값을 가져오기 위한 조회 -- (2)

하지만, select_related를 사용한 2번의 경우 DB에 1번만 접근한다. select_related에서 관련된 Object(여기서는 city)까지 한 번에 조회해 cache에 저장한다. 따라서, 이후에 city 값에 대한 조회를 할 때는 DB에 재접근 방식이 아닌 가져온 cache 값에서 꺼내서 사용하기만 하면 된다.

➰ select_related는 INNER JOIN을 통해 데이터를 가져온다. ➰ foreign-key, one-to-one과 같은 1:1 관계에서 사용하는 것이 효과적이다.

👉 prefetch_related : 정참조, 역참조 모두

정참조를 포함하여, 역참조(다른 객체가 ForeignKey를 가지고 있거나 M:N 관계일 때, 해당 객체를 참조하고 있는 다른 객체를 참조)하는 경우 사용한다.

➰ prefetch_related는 foreign-key, one-to-one 뿐만 아니라 many-to-many, many-to-one 관계에서도 사용 가능하다. ➰ SQL에서 각 모델을 조회하고, Python에서 Join을 실행한다.

간단한 테이블 및 구조로 확인해 보았지만, 데이터가 많아지고 연관된 테이블이 많아질 경우 DB 접근 방식에 따른 쿼리 성능은 무시할 수 없다.

상황에 맞게 select_related와 prefetch_related를 사용해 효율적인 코딩을 해보자!😆

• 필요에 따라 git config 세팅 git config (–global) user.name “[user-name]” git config (–global) user.email [email@address.com]

• git 저장소 로컬로 복제 git clone https://github.com/.git (public) git clone https://[user-name]@github.com/.git (private)

• 로컬 <-> 원격 저장소 연결 git remote -v git remote add origin github.com/*.git

• 브랜치 생성 후 이동 git checkout -b [branch]

• 브랜치 삭제 git branch -d [branch]

• 원격 저장소 pull git pull origin [branch]

• 변경사항 저장 git commit -m '[commit message]' -m: 옵션 (커밋 메세지 지정)

• 원격 저장소 push git push origin [branch]

• 현재 브랜치에 다른 브랜치 수정사항 병합 git merge [another branch]

예를 들어, 우리가 작업한 프로젝트에 버그가 있어 수정을 해야 하는 상황이라고 가정하자. 버그 하나를 잡기 위해 여러 개의 파일을 수정하며 해당 기능과 연관된 모든 파일을 수정해야 한다면 그것은 클린한 코드일까? 시간과 함께 효율성도 매우 떨어지는 작업이 될 것 같다. 위와 같이 작업되어 있다면 아마 코드의 재사용성도 그리 높진 않을 것이다. 이러한 점들을 해결하고, 보다 효율적인 프로그래밍을 위해 클린 아키텍처 패턴을 적용하도록 한다.

Clean Architecture : 클린 아키텍처

이미지 출처: http://blog.cleancoder.com/uncle-bob/2012/08/13/the-clean-architecture.html

Clean + Architecture : 깔끔한 구조 말 그대로 깔끔한 구조의 디자인 패턴이며, 위 그림의 화살표 방향(External Interfaces -> Infrastructure -> Use Case -> Entity)은 의존성(Dependency)의 방향을 의미한다.

그림을 안쪽부터 순서대로 보면 다음과 같다.

1. Entity : 엔티티

• 핵심 업무 규칙을 캡슐화한다.
• 변형이 가장 적고, 외부의 영향을 최소화하는 영역이다.

2. Use Case : 유스케이스

• 어플리케이션의 비즈니스 규칙이 포함될 수 있다.
• 사용자의 요청에 알맞게 상위 계층(Entity)를 가공, 출력한다.

3. Infrastructure : 인터페이스 어댑터(컨트롤러, 게이트웨이, 프레젠터)

• 데이터 형식을 변환하는 역할이다.
• 외부 플랫폼에 따라 내부 비즈니스 로직은 유지한 채 어댑터의 로직 일부만 수정해 편리하게 개발 가능하도록 도와준다.

4. External Interfaces : 외부 영역 (DB, API, UI, Devices..)

• 웹 혹은 앱의 프레임워크로서, 내부 비즈니스 로직과 직접적인 영향은 없는 영역이다.
• 언제든 변경이 가능하다.

📌 핵심은 내부이다. 📌 내부 영역으로 갈수록 변경이 적어야 한다. 📌 외부에서는 내부 영역에 의존할 수 있지만, 내부에서는 외부 영역으로의 의존성이 적용될 수 없다. ex) Use Case는 Entity를 사용할 수 있어도, Entity는 Use Case를 사용 불가

DDD(Domain-Driven Design) : 도메인 주도 설계

도메인을 중심으로 설계해 나가는 방법이다. 여기서의 '도메인'은 사용자가 필요한 모든 비즈니스 영역, 업무의 집합이다. 주문 사이트를 예로 들었을 때, 주문, 배송, 마케팅은 각각의 도메인에 해당한다고 볼 수 있다. 이러한 도메인들이 서로 상호작용을 하며 설계하는 방식을 도메인 주도 설계라고 한다.

📌 각각의 도메인은 서로 철저히 분리된다. 📌 핵심은 "Loose Coupling, High Cohesion", 높은 응집력과 낮은 결합도로 변경과 확장에 용이한 설계를 얻게된다.

정리하면, 도메인을 각 서비스별로 분리하는게 핵심이다. 하지만 작업자에 따라, 사용자의 보는 관점에 따라 도메인은 달라질 수 있다. 따라서 정해진 답은 없다.