20분정도에 말씀 하신 내용 질문 있습니다.

마이크로서비스 아키텍처 와 MSA 패턴 이해

마이크로 프런트엔드, 통신 패턴

해결된 질문

작성

수정됨

카프카에 최종 참여자 지원 LPS라는 설정이라는걸 구글링 해봤는데 잘 보이지가 않아서 혹시 어떤 설정이 있어야되나요?

말씀하신 내용으론 큐잉 서비스에서 최종 데이터베이스에 정착에 되는 응답값을 확인후 큐 서비스에서 지워지는 기능이 아주 매력적으로 느껴져서 여쭤봅니다

질문이 많아서 죄송합니다, 회사 프로젝트에 리더급으로 참여를 하여 궁금한걸 전부 물어보게 되네요;;ㅠㅠ

감사합니다.

아키텍처 msa

답변 2

han jeong heon

지식공유자

움 아키텍처를 설명하기 위해 개념적인 용어들을 사용했지만 이런 기능은 카프카 매커니즘이 이미 제공하고 있습니다. 예를 들어 간단히 구현해 보면 다음과 같습니다.

주문 서비스가 새로운 주문을 생성하면, 이 주문 정보를 Kafka에 이벤트로 보내고, 주문 이력 서비스가 이 이벤트를 읽어 데이터베이스에 주문 이력을 저장한 후, 메시지가 성공적으로 처리되었음을 확인하고 Kafka에서 해당 메시지를 제거하는 방식으로 구현할 수 있습니다.

아래는 이 과정을 간단한 샘플 코드와 함께 설명하겠습니다.

1. Kafka에 주문 이벤트 보내기 (주문 서비스)

주문 서비스가 주문 정보를 Kafka에 보내는 과정입니다. Python의 kafka-python 라이브러리를 사용하여 Kafka에 메시지를 보내는 예제입니다.

from kafka import KafkaProducer 
import json 
# Kafka 프로듀서 설정 
producer = KafkaProducer(
 bootstrap_servers='localhost:9092', 
 value_serializer=lambda v: json.dumps(v).encode('utf-8') 
) 
# 주문 데이터 
order = {
 'order_id': '12345', 
 'customer_id': '67890', 
  'amount': 100.0 } 
# Kafka 토픽에 주문 이벤트 전송 
producer.send('order-topic', order) 
producer.flush()

2. 주문 이력 저장 및 Kafka 메시지 삭제 (주문 이력 서비스)

주문 이력 서비스가 Kafka에서 메시지를 읽어 데이터베이스에 저장하고, 저장이 완료되면 Kafka에서 해당 메시지를 제거합니다. Python의 kafka-python과 mysql-connector-python을 사용한 예제입니다.

from kafka import KafkaConsumer, KafkaProducer
import mysql.connector
import json

# Kafka 컨슈머 설정
consumer = KafkaConsumer(
    'order-topic',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    group_id='order-history-group',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

# Kafka 프로듀서 설정
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# MySQL 데이터베이스 설정
db_conn = mysql.connector.connect(
    host='localhost',
    user='youruser',
    password='yourpassword',
    database='yourdatabase'
)
cursor = db_conn.cursor()

# Kafka 메시지 처리
for message in consumer:
    order = message.value
    order_id = order['order_id']
    customer_id = order['customer_id']
    amount = order['amount']
    
    # 주문 이력을 데이터베이스에 저장
    cursor.execute(
        "INSERT INTO OrderHistory (order_id, customer_id, amount) VALUES (%s, %s, %s)",
        (order_id, customer_id, amount)
    )
    db_conn.commit()

    # 메시지 처리가 완료된 후 Kafka에서 메시지 삭제 (offset 커밋)
    consumer.commit()
    
    # 메시지 전송 상태를 확인하고 로그 기록 또는 다른 작업 수행
    print(f"Order {order_id} processed and saved to database.")

코드 복사

요약

주문 서비스: 주문을 Kafka 토픽에 메시지로 전송합니다.
주문 이력 서비스:
- Kafka에서 메시지를 읽어 데이터베이스에 주문 이력을 저장합니다.
- 메시지를 처리한 후 Kafka의 오프셋을 커밋하여 메시지가 더 이상 처리되지 않도록 합니다.
- 엄밀히 말하면 Kafka에서 오프셋(offset)은 특정 파티션 내에서 메시지의 위치를 나타냅니다. 즉 소비자가 메시지를 처리한 후, Kafka의 컨슈머 그룹은 해당 메시지가 성공적으로 처리되었다는 것을 나타내기 위해 오프셋을 커밋합니다.
- 오프셋 커밋은 Kafka가 메시지를 어느 지점까지 읽었는지를 기록합니다. 즉, 컨슈머가 다음에 재시작될 때, 마지막으로 커밋된 오프셋부터 메시지를 읽기 시작합니다. 따라서 한번 전달된 메시지가 다시 전달되지 않음을 보장합니다. 그리고 카프카의 메시지는 카프카 로그 보존기간에 따라 자동으로 삭제되게 하면 됩니다.

최강개발자

질문자

아 단순하게 데이터베이스 insert 완료후 메시지를 삭제하는 메커니즘을 말씀하신거군요 감사합니다.

인프런 AI 인턴

안녕하세요, 인프런 AI 인턴입니다. 최강개발자님.

카프카에 대해 질문해주신 내용과 유사한 기존 답변을 찾아보니, 카프카의 사용과 관련된 설정 및 질문에 대한 답변이 있었습니다. 해당 내용을 참고하시면 도움이 될 것 같습니다.

추가 설명:

카프카의 exactly-once semantics 설정을 통해 큐잉 서비스에서 데이터베이스로의 정착과 그에 따른 메시지 삭제를 트랜잭셔널하게 처리할 수 있습니다. 예를 들어, 다음과 같은 설정이 가능합니다:
1. enable.idempotence=true로 설정하면 프로듀서가 중복 없이 메시지를 전송합니다.
2. transactional.id를 사용하여 트랜잭션을 정의하고, 소비자가 이 트랜잭션 내에서 메시지를 읽고 처리하여 데이터베이스에 기록하는 방식으로 운영할 수 있습니다.

이렇게 하면 데이터베이스 기록 후 큐에서 안전하게 메시지를 지울 수 있습니다.

저는 질문자님의 이해를 돕고자 지식공유자님의 기존 답변을 참고하여 유사한 답변 링크를 추천해드리고 있습니다.
현재 베타 기능으로 답변이 만족스럽지 않을 수 있는 점 양해 부탁드립니다. 🙏
추가적으로 궁금한 점이 있으시면, 이어서 질문해 주세요. 곧 지식공유자께서 답변해 주실 것입니다.

인프런 커뮤니티 질문&답변