1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# docker-compose.yml
version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      
  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// Kafka 생산자 예제
import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class OrderEventProducer {
    public static void main(String[] args) {
        // 생산자 설정
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        // 생산자 인스턴스 생성
        Producer<String, String> producer = new KafkaProducer<>(props);
        
        // 주문 이벤트 메시지 생성
        String orderId = "ORD-" + System.currentTimeMillis();
        String orderData = "{\"orderId\":\"" + orderId + "\",\"customerId\":\"CUST001\",\"amount\":150.00}";
        
        // 메시지 전송
        ProducerRecord<String, String> record = 
            new ProducerRecord<>("order-events", orderId, orderData);
            
        // 비동기 전송 (콜백 포함)
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e != null) {
                    // 에러 처리
                    System.err.println("메시지 전송 실패: " + e.getMessage());
                } else {
                    // 성공 로깅
                    System.out.println("메시지 전송 성공: " + 
                        "토픽=" + metadata.topic() + ", " +
                        "파티션=" + metadata.partition() + ", " +
                        "오프셋=" + metadata.offset());
                }
            }
        });
        
        // 리소스 정리
        producer.flush();
        producer.close();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

// Kafka 소비자 예제
import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class OrderEventConsumer {
    public static void main(String[] args) {
        // 소비자 설정
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "order-processing-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest"); // 처음부터 읽기
        
        // 소비자 인스턴스 생성
        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        
        // 토픽 구독
        consumer.subscribe(Arrays.asList("order-events"));
        
        // 주기적으로 데이터 폴링
        try {
            while (true) {
                // 100ms 동안 데이터 기다림
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    // 메시지 처리
                    System.out.println("수신 데이터: " + 
                        "키=" + record.key() + ", " +
                        "값=" + record.value() + ", " +
                        "토픽=" + record.topic() + ", " +
                        "파티션=" + record.partition() + ", " +
                        "오프셋=" + record.offset());
                    
                    // 여기서 주문 처리 로직 구현
                    processOrder(record.value());
                }
                
                // 명시적 커밋 (자동 커밋을 사용할 수도 있음)
                consumer.commitSync();
            }
        } finally {
            // 리소스 정리
            consumer.close();
        }
    }
    
    private static void processOrder(String orderJson) {
        // 주문 처리 로직
        System.out.println("주문 처리 중: " + orderJson);
    }
}

1
2
3
4
5

dependencies {
    implementation 'org.springframework.kafka:spring-kafka'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    // 기타 필요한 의존성
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      group-id: order-processing-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: com.example.model

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

package com.example.model;

import java.math.BigDecimal;
import java.time.LocalDateTime;

public class OrderEvent {
    private String orderId;
    private String customerId;
    private BigDecimal amount;
    private LocalDateTime timestamp;
    
    // 생성자, 게터, 세터 생략
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

package com.example.service;

import com.example.model.OrderEvent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.time.LocalDateTime;

@Service
public class OrderEventProducer {
    
    private static final String TOPIC = "order-events";
    
    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;
    
    public void sendOrderEvent(OrderEvent orderEvent) {
        // 타임스탬프 설정
        orderEvent.setTimestamp(LocalDateTime.now());
        
        // 키는 주문 ID, 값은 OrderEvent 객체
        ListenableFuture<SendResult<String, OrderEvent>> future = 
            kafkaTemplate.send(TOPIC, orderEvent.getOrderId(), orderEvent);
            
        future.addCallback(new ListenableFutureCallback<SendResult<String, OrderEvent>>() {
            @Override
            public void onSuccess(SendResult<String, OrderEvent> result) {
                System.out.println("주문 이벤트 전송 성공: " + orderEvent.getOrderId() + 
                    ", 오프셋: " + result.getRecordMetadata().offset());
            }
            
            @Override
            public void onFailure(Throwable ex) {
                System.err.println("주문 이벤트 전송 실패: " + orderEvent.getOrderId() + 
                    ", 원인: " + ex.getMessage());
            }
        });
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

package com.example.service;

import com.example.model.OrderEvent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class OrderEventConsumer {
    
    @KafkaListener(topics = "order-events", groupId = "order-processing-group")
    public void consume(OrderEvent orderEvent) {
        System.out.println("주문 이벤트 수신: " + orderEvent.getOrderId());
        
        // 여기서 비즈니스 로직 처리
        processOrder(orderEvent);
    }
    
    private void processOrder(OrderEvent orderEvent) {
        // 주문 처리 로직
        System.out.println("주문 처리 중: " + orderEvent.getOrderId() + 
            ", 고객: " + orderEvent.getCustomerId() + 
            ", 금액: " + orderEvent.getAmount());
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

package com.example.streaming;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Properties;

public class OrderProcessingStream {
    public static void main(String[] args) {
        // 스트림 설정
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-analysis-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        
        // 스트림 빌더
        StreamsBuilder builder = new StreamsBuilder();
        
        // 입력 스트림 생성
        KStream<String, String> orderEvents = builder.stream("order-events");
        
        // 주문 금액으로 필터링 (JSON 문자열 파싱 가정)
        KStream<String, String> highValueOrders = orderEvents
            .filter((key, value) -> {
                // JSON 파싱 및 금액 추출 로직
                double amount = extractAmountFromJson(value);
                return amount > 1000.0; // 고액 주문 필터링
            });
        
        // 고액 주문을 별도 토픽으로 전송
        highValueOrders.to("high-value-orders", Produced.with(Serdes.String(), Serdes.String()));
        
        // 고객별 주문 합계 계산
        KTable<String, Double> customerOrderTotals = orderEvents
            .groupBy((key, value) -> extractCustomerIdFromJson(value))
            .aggregate(
                () -> 0.0, // 초기값
                (customerId, orderJson, total) -> {
                    double amount = extractAmountFromJson(orderJson);
                    return total + amount;
                }
            );
        
        // 고객별 합계를 별도 토픽으로 전송
        customerOrderTotals.toStream().to("customer-order-totals", 
            Produced.with(Serdes.String(), Serdes.Double()));
        
        // 스트림 생성 및 시작
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
        
        // 종료 처리
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
    
    // JSON 문자열에서 금액 추출 (간단한 예시)
    private static double extractAmountFromJson(String json) {
        // 실제 구현에서는 JSON 라이브러리 사용
        // 여기서는 예시로 간단히 구현
        if (json.contains("\"amount\":")) {
            int start = json.indexOf("\"amount\":") + "\"amount\":".length();
            int end = json.indexOf(",", start);
            if (end == -1) {
                end = json.indexOf("}", start);
            }
            return Double.parseDouble(json.substring(start, end).trim());
        }
        return 0.0;
    }
    
    // JSON 문자열에서 고객 ID 추출
    private static String extractCustomerIdFromJson(String json) {
        // 실제 구현에서는 JSON 라이브러리 사용
        if (json.contains("\"customerId\":")) {
            int start = json.indexOf("\"customerId\":") + "\"customerId\":".length();
            if (json.charAt(start) == '\"') {
                start++; // 따옴표 건너뛰기
                int end = json.indexOf("\"", start);
                return json.substring(start, end);
            }
        }
        return "unknown";
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

// 이벤트 클래스
public abstract class DomainEvent {
    private final String aggregateId;
    private final long timestamp;
    
    public DomainEvent(String aggregateId) {
        this.aggregateId = aggregateId;
        this.timestamp = System.currentTimeMillis();
    }
    
    // 게터 메소드
}

// 주문 생성 이벤트
public class OrderCreatedEvent extends DomainEvent {
    private final String customerId;
    private final BigDecimal initialAmount;
    
    public OrderCreatedEvent(String orderId, String customerId, BigDecimal initialAmount) {
        super(orderId);
        this.customerId = customerId;
        this.initialAmount = initialAmount;
    }
    
    // 게터 메소드
}

// 주문 항목 추가 이벤트
public class OrderItemAddedEvent extends DomainEvent {
    private final String productId;
    private final int quantity;
    private final BigDecimal price;
    
    public OrderItemAddedEvent(String orderId, String productId, int quantity, BigDecimal price) {
        super(orderId);
        this.productId = productId;
        this.quantity = quantity;
        this.price = price;
    }
    
    // 게터 메소드
}

// 이벤트 저장소 인터페이스
public interface EventStore {
    void saveEvents(String aggregateId, List<DomainEvent> events, int expectedVersion);
    List<DomainEvent> getEvents(String aggregateId);
}

// 주문 엔티티(애그리게이트 루트)
public class Order {
    private String orderId;
    private String customerId;
    private BigDecimal totalAmount;
    private List<OrderItem> items = new ArrayList<>();
    private int version = -1;
    private List<DomainEvent> changes = new ArrayList<>();
    
    // 새 주문 생성
    public static Order createOrder(String orderId, String customerId) {
        Order order = new Order();
        order.applyChange(new OrderCreatedEvent(orderId, customerId, BigDecimal.ZERO));
        return order;
    }
    
    // 이벤트 기록에서 주문 로드
    public static Order loadFromHistory(List<DomainEvent> history) {
        Order order = new Order();
        history.forEach(order::applyChange);
        return order;
    }
    
    // 항목 추가
    public void addItem(String productId, int quantity, BigDecimal price) {
        applyChange(new OrderItemAddedEvent(orderId, productId, quantity, price));
    }
    
    // 변경 사항 적용 및 기록
    private void applyChange(DomainEvent event) {
        applyEvent(event);
        changes.add(event);
    }
    
    // 이벤트 적용
    private void applyEvent(DomainEvent event) {
        if (event instanceof OrderCreatedEvent) {
            OrderCreatedEvent e = (OrderCreatedEvent) event;
            this.orderId = e.getAggregateId();
            this.customerId = e.getCustomerId();
            this.totalAmount = e.getInitialAmount();
            this.version++;
        } else if (event instanceof OrderItemAddedEvent) {
            OrderItemAddedEvent e = (OrderItemAddedEvent) event;
            OrderItem item = new OrderItem(e.getProductId(), e.getQuantity(), e.getPrice());
            this.items.add(item);
            this.totalAmount = this.totalAmount.add(item.getTotalPrice());
            this.version++;
        }
    }
    
    // 변경 사항 커밋
    public List<DomainEvent> getUncommittedChanges() {
        return new ArrayList<>(changes);
    }
    
    public void markChangesAsCommitted() {
        changes.clear();
    }
    
    // 게터 메소드
}

// 주문 서비스
public class OrderService {
    private final EventStore eventStore;
    
    public OrderService(EventStore eventStore) {
        this.eventStore = eventStore;
    }
    
    public String createOrder(String customerId) {
        String orderId = UUID.randomUUID().toString();
        Order order = Order.createOrder(orderId, customerId);
        eventStore.saveEvents(orderId, order.getUncommittedChanges(), -1);
        order.markChangesAsCommitted();
        return orderId;
    }
    
    public void addOrderItem(String orderId, String productId, int quantity, BigDecimal price) {
        List<DomainEvent> events = eventStore.getEvents(orderId);
        Order order = Order.loadFromHistory(events);
        order.addItem(productId, quantity, price);
        eventStore.saveEvents(orderId, order.getUncommittedChanges(), order.getVersion() - order.getUncommittedChanges().size());
        order.markChangesAsCommitted();
    }
    
    public Order getOrder(String orderId) {
        List<DomainEvent> events = eventStore.getEvents(orderId);
        return Order.loadFromHistory(events);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 브로커 측 설정
# 데이터 손실 방지를 위한 설정
min.insync.replicas=2
unclean.leader.election.enable=false

# 프로듀서 측 설정
acks=all        # 모든 복제본 확인
retries=10      # 재시도 횟수
max.in.flight.requests.per.connection=1  # 순서 보장
enable.idempotence=true  # 멱등성 활성화

# 트랜잭션 사용 (정확히 한 번 처리)
transactional.id=unique-id-per-producer

1
2
3
4
5
6
7
8

# MirrorMaker 2 설정
clusters=source, target
source.bootstrap.servers=source-kafka:9092
target.bootstrap.servers=target-kafka:9092

# 소스 -> 타겟 복제 설정
source->target.enabled=true
source->target.topics=.*  # 모든 토픽 복제

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

// 소스 커넥터 설정 (데이터베이스 변경 캡처)
{
  "name": "mysql-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-server",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "1",
    "database.server.name": "mysql-server-1",
    "table.whitelist": "inventory.customers",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "dbhistory.inventory"
  }
}

// 싱크 커넥터 설정 (ElasticSearch로 데이터 전송)
{
  "name": "elasticsearch-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "mysql-server-1.inventory.customers",
    "connection.url": "http://elasticsearch:9200",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "key.ignore": "false",
    "schema.ignore": "true",
    "type.name": "customer"
  }
}