먼저 라이브러리를 설치해야겠지?

pip install kafka-python

Docker로 카프카 실행하기 (간단 버전)

docker run -d --name kafka -p 9092:9092 \
  -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  wurstmeister/kafka

from kafka import KafkaProducer
import json
import time
from datetime import datetime

# 프로듀서 생성
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],  # 카프카 서버 주소
    value_serializer=lambda v: json.dumps(v).encode('utf-8')  # JSON으로 직렬화
)

# 사용자 클릭 이벤트 전송
def send_click_event(user_id, page, action):
    event = {
        'user_id': user_id,
        'page': page,
        'action': action,
        'timestamp': datetime.now().isoformat()
    }
    
    # 'user-clicks' 토픽으로 전송
    producer.send('user-clicks', value=event)
    print(f"이벤트 전송 완료: {event}")

# 테스트: 10개의 클릭 이벤트 전송
for i in range(10):
    send_click_event(
        user_id=f"user_{i % 3}",  # user_0, user_1, user_2 순환
        page=f"/page/{i % 5}",
        action="click"
    )
    time.sleep(1)  # 1초 대기

# 모든 메시지가 전송될 때까지 대기
producer.flush()
print("모든 이벤트 전송 완료!")

from kafka import KafkaConsumer
import json

# 컨슈머 생성
consumer = KafkaConsumer(
    'user-clicks',  # 구독할 토픽
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',  # 처음부터 읽기
    enable_auto_commit=True,  # 자동으로 오프셋 커밋
    group_id='click-analyzer-group',  # 컨슈머 그룹 ID
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

print("클릭 이벤트 수신 대기 중...")

# 메시지 수신 및 처리
for message in consumer:
    event = message.value
    
    print(f"\n[새 이벤트 수신]")
    print(f"사용자: {event['user_id']}")
    print(f"페이지: {event['page']}")
    print(f"액션: {event['action']}")
    print(f"시간: {event['timestamp']}")
    
    # 여기서 실제 비즈니스 로직 처리
    # 예: 데이터베이스 저장, 분석, 알림 전송 등
    
    # 특정 조건에 따른 처리 예시
    if event['action'] == 'click' and '/page/3' in event['page']:
        print("⚠️ 중요 페이지 클릭 감지!")

from kafka import KafkaProducer
import json
import random
from datetime import datetime

class OrderProducer:
    def __init__(self):
        self.producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
    
    def create_order(self, user_id, product_id, quantity, price):
        order = {
            'order_id': f"ORD-{datetime.now().strftime('%Y%m%d%H%M%S')}-{random.randint(1000, 9999)}",
            'user_id': user_id,
            'product_id': product_id,
            'quantity': quantity,
            'price': price,
            'total_amount': quantity * price,
            'status': 'created',
            'timestamp': datetime.now().isoformat()
        }
        
        # orders 토픽으로 전송
        self.producer.send('orders', value=order)
        print(f"✅ 주문 생성: {order['order_id']} - {order['product_id']} x {quantity}")
        
        return order
    
    def close(self):
        self.producer.flush()
        self.producer.close()

# 사용 예시
if __name__ == "__main__":
    order_producer = OrderProducer()
    
    # 테스트 주문 생성
    products = [
        ('PROD-001', '노트북', 1500000),
        ('PROD-002', '마우스', 30000),
        ('PROD-003', '키보드', 80000),
        ('PROD-004', '모니터', 350000)
    ]
    
    for i in range(5):
        product = random.choice(products)
        order_producer.create_order(
            user_id=f"USER-{random.randint(1, 100)}",
            product_id=product[0],
            quantity=random.randint(1, 3),
            price=product[2]
        )
    
    order_producer.close()

from kafka import KafkaConsumer, KafkaProducer
import json

class InventoryChecker:
    def __init__(self):
        self.consumer = KafkaConsumer(
            'orders',
            bootstrap_servers=['localhost:9092'],
            auto_offset_reset='latest',
            group_id='inventory-checker-group',
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
        
        self.producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        
        # 간단한 재고 DB (실제로는 데이터베이스 사용)
        self.inventory = {
            'PROD-001': 10,
            'PROD-002': 50,
            'PROD-003': 30,
            'PROD-004': 15
        }
    
    def check_inventory(self, order):
        product_id = order['product_id']
        quantity = order['quantity']
        
        if product_id in self.inventory and self.inventory[product_id] >= quantity:
            # 재고 충분
            self.inventory[product_id] -= quantity
            order['status'] = 'inventory_confirmed'
            order['inventory_available'] = True
            
            # payment 토픽으로 전달
            self.producer.send('payment', value=order)
            print(f"✅ 재고 확인 완료: {order['order_id']} - 남은 재고: {self.inventory[product_id]}")
        else:
            # 재고 부족
            order['status'] = 'inventory_failed'
            order['inventory_available'] = False
            
            # notifications 토픽으로 전달
            self.producer.send('notifications', value={
                'type': 'inventory_shortage',
                'order': order,
                'message': '죄송합니다. 재고가 부족합니다.'
            })
            print(f"❌ 재고 부족: {order['order_id']}")
    
    def start(self):
        print("재고 확인 서비스 시작...")
        for message in self.consumer:
            order = message.value
            self.check_inventory(order)

if __name__ == "__main__":
    checker = InventoryChecker()
    checker.start()

from kafka import KafkaConsumer, KafkaProducer
import json
import random
import time

class PaymentProcessor:
    def __init__(self):
        self.consumer = KafkaConsumer(
            'payment',
            bootstrap_servers=['localhost:9092'],
            auto_offset_reset='latest',
            group_id='payment-processor-group',
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
        
        self.producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
    
    def process_payment(self, order):
        print(f"💳 결제 처리 중: {order['order_id']} - {order['total_amount']:,}원")
        
        # 결제 처리 시뮬레이션 (실제로는 PG사 API 호출)
        time.sleep(1)
        
        # 90% 확률로 성공
        if random.random() < 0.9:
            order['status'] = 'payment_completed'
            order['payment_success'] = True
            
            # shipping 토픽으로 전달
            self.producer.send('shipping', value=order)
            
            # 고객에게 알림
            self.producer.send('notifications', value={
                'type': 'payment_success',
                'order': order,
                'message': f"결제가 완료되었습니다. 금액: {order['total_amount']:,}원"
            })
            
            print(f"✅ 결제 성공: {order['order_id']}")
        else:
            order['status'] = 'payment_failed'
            order['payment_success'] = False
            
            # 고객에게 알림
            self.producer.send('notifications', value={
                'type': 'payment_failed',
                'order': order,
                'message': '결제에 실패했습니다. 다시 시도해주세요.'
            })
            
            print(f"❌ 결제 실패: {order['order_id']}")
    
    def start(self):
        print("결제 처리 서비스 시작...")
        for message in self.consumer:
            order = message.value
            self.process_payment(order)

if __name__ == "__main__":
    processor = PaymentProcessor()
    processor.start()

# 키를 지정해서 전송
producer.send(
    'user-events',
    key=user_id.encode('utf-8'),  # 키 지정
    value=event
)

# 같은 user_id는 항상 같은 파티션으로!
# 이렇게 하면 사용자별 이벤트 순서가 보장돼

# 수동 커밋 예제
consumer = KafkaConsumer(
    'orders',
    enable_auto_commit=False,  # 자동 커밋 비활성화
    group_id='order-processor'
)

for message in consumer:
    try:
        # 메시지 처리
        process_order(message.value)
        
        # 데이터베이스 저장
        save_to_db(message.value)
        
        # 처리 완료 후 커밋
        consumer.commit()
        print(f"✅ 오프셋 커밋: {message.offset}")
        
    except Exception as e:
        print(f"❌ 처리 실패: {e}")
        # 커밋하지 않으면 다음에 다시 처리됨!

# Exactly Once 설정
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    acks='all',  # 모든 복제본이 확인할 때까지 대기
    retries=3,  # 실패 시 재시도
    enable_idempotence=True,  # 멱등성 활성화 (중복 방지)
    transactional_id='order-producer-1'  # 트랜잭션 ID
)

# 트랜잭션 시작
producer.init_transactions()
producer.begin_transaction()

try:
    # 여러 메시지를 원자적으로 전송
    producer.send('orders', order1)
    producer.send('inventory', inventory_update)
    producer.send('notifications', notification)
    
    # 모두 성공하면 커밋
    producer.commit_transaction()
    print("✅ 트랜잭션 커밋 완료")
    
except Exception as e:
    # 하나라도 실패하면 롤백
    producer.abort_transaction()
    print(f"❌ 트랜잭션 롤백: {e}")

from kafka import KafkaConsumer, TopicPartition
from kafka.structs import OffsetAndMetadata

def check_consumer_lag(topic, group_id):
    consumer = KafkaConsumer(
        bootstrap_servers=['localhost:9092'],
        group_id=group_id
    )
    
    # 토픽의 모든 파티션 가져오기
    partitions = consumer.partitions_for_topic(topic)
    
    total_lag = 0
    
    for partition in partitions:
        tp = TopicPartition(topic, partition)
        
        # 현재 오프셋 (컨슈머가 읽은 위치)
        committed = consumer.committed(tp)
        current_offset = committed if committed else 0
        
        # 최신 오프셋 (프로듀서가 쓴 위치)
        consumer.seek_to_end(tp)
        end_offset = consumer.position(tp)
        
        # Lag 계산
        lag = end_offset - current_offset
        total_lag += lag
        
        print(f"파티션 {partition}: Lag = {lag:,} 메시지")
    
    print(f"\n총 Lag: {total_lag:,} 메시지")
    
    # 경고 임계값
    if total_lag > 10000:
        print("⚠️ 경고: Consumer Lag이 너무 높습니다!")
        print("→ 컨슈머 인스턴스를 추가하세요.")
    elif total_lag > 1000:
        print("⚡ 주의: Consumer Lag이 증가하고 있습니다.")
    else:
        print("✅ 정상: Consumer Lag이 안정적입니다.")
    
    consumer.close()

# 사용 예시
check_consumer_lag('orders', 'order-processor-group')

# 최적화된 프로듀서 설정
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    
    # 배치 처리
    batch_size=32768,  # 32KB
    linger_ms=10,  # 10ms 대기
    
    # 압축
    compression_type='snappy',
    
    # 버퍼
    buffer_memory=67108864,  # 64MB
    
    # 재시도
    retries=3,
    retry_backoff_ms=100,
    
    # 성능
    acks='1',  # 리더만 확인 (빠름)
    max_in_flight_requests_per_connection=5
)

from kafka import KafkaConsumer, KafkaProducer
import json
import redis
from datetime import datetime

class RealTimeInventoryManager:
    def __init__(self):
        # 카프카 컨슈머 (주문 이벤트 수신)
        self.consumer = KafkaConsumer(
            'order-events',
            bootstrap_servers=['localhost:9092'],
            group_id='inventory-manager',
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
        
        # 카프카 프로듀서 (재고 업데이트 알림)
        self.producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        
        # Redis (실시간 재고 캐시)
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
    
    def get_current_stock(self, product_id):
        """현재 재고 조회"""
        stock = self.redis_client.get(f"stock:{product_id}")
        return int(stock) if stock else 0
    
    def update_stock(self, product_id, quantity_change):
        """재고 업데이트"""
        current = self.get_current_stock(product_id)
        new_stock = current + quantity_change
        
        # Redis 업데이트
        self.redis_client.set(f"stock:{product_id}", new_stock)
        
        # 재고 변경 이벤트 발행
        event = {
            'product_id': product_id,
            'previous_stock': current,
            'new_stock': new_stock,
            'change': quantity_change,
            'timestamp': datetime.now().isoformat()
        }
        
        self.producer.send('inventory-updates', value=event)
        
        # 재고 부족 경고
        if new_stock < 10:
            self.producer.send('inventory-alerts', value={
                'type': 'low_stock',
                'product_id': product_id,
                'current_stock': new_stock,
                'message': f'재고 부족 경고! 현재 재고: {new_stock}개'
            })
            print(f"⚠️ 재고 부족 경고: {product_id} - {new_stock}개")
        
        return new_stock
    
    def process_order(self, order):
        """주문 처리"""
        product_id = order['product_id']
        quantity = order['quantity']
        
        print(f"\n📦 주문 처리: {order['order_id']}")
        print(f"   상품: {product_id}, 수량: {quantity}")
        
        # 재고 차감
        new_stock = self.update_stock(product_id, -quantity)
        print(f"   재고 업데이트: {new_stock}개 남음")
        
        # 실시간 대시보드 업데이트
        self.producer.send('dashboard-updates', value={
            'type': 'stock_change',
            'product_id': product_id,
            'stock': new_stock,
            'timestamp': datetime.now().isoformat()
        })
    
    def start(self):
        """서비스 시작"""
        print("🚀 실시간 재고 관리 시스템 시작...")
        print("주문 이벤트 대기 중...\n")
        
        for message in self.consumer:
            order = message.value
            
            if order['type'] == 'order_created':
                self.process_order(order)
            elif order['type'] == 'order_cancelled':
                # 주문 취소 시 재고 복구
                self.update_stock(order['product_id'], order['quantity'])
                print(f"↩️ 주문 취소: 재고 복구 완료")

# 실행
if __name__ == "__main__":
    manager = RealTimeInventoryManager()
    manager.start()

# 용량 계산 예시

# 1. 메시지 처리량 계산
messages_per_second = 10000  # 초당 메시지 수
message_size_bytes = 1024  # 메시지 크기 (1KB)
throughput_mb_per_sec = (messages_per_second * message_size_bytes) / (1024 * 1024)

print(f"필요한 처리량: {throughput_mb_per_sec:.2f} MB/s")

# 2. 저장 용량 계산
retention_hours = 168  # 7일 보관
total_storage_gb = (throughput_mb_per_sec * 3600 * retention_hours) / 1024

print(f"필요한 저장 용량: {total_storage_gb:.2f} GB")

# 3. 복제 고려
replication_factor = 3  # 3개 복제
total_storage_with_replication = total_storage_gb * replication_factor

print(f"복제 포함 저장 용량: {total_storage_with_replication:.2f} GB")

# 4. 여유분 추가 (50%)
recommended_storage = total_storage_with_replication * 1.5

print(f"권장 저장 용량: {recommended_storage:.2f} GB")

# 출력 예시:
# 필요한 처리량: 9.77 MB/s
# 필요한 저장 용량: 5908.20 GB
# 복제 포함 저장 용량: 17724.61 GB
# 권장 저장 용량: 26586.91 GB (약 26TB)