Kafka

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# Kafka 다운로드
wget https://downloads.apache.org/kafka/3.4.0/kafka_2.13-3.4.0.tgz
tar -xzf kafka_2.13-3.4.0.tgz
cd kafka_2.13-3.4.0

# ZooKeeper 시작
bin/zookeeper-server-start.sh config/zookeeper.properties &

# Kafka 브로커 시작
bin/kafka-server-start.sh config/server.properties &

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

# 첫 번째 브로커
broker.id=0
listeners=PLAINTEXT://server1:9092
log.dirs=/var/lib/kafka/data

# 두 번째 브로커
broker.id=1
listeners=PLAINTEXT://server2:9092
log.dirs=/var/lib/kafka/data

# 세 번째 브로커
broker.id=2
listeners=PLAINTEXT://server3:9092
log.dirs=/var/lib/kafka/data

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

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      
  kafka:
    image: confluentinc/cp-kafka:7.3.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

1
2
3
4
5

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.4.0</version>
</dependency>

 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

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class SimpleProducer {
    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);
        
        // 메시지 전송
        for (int i = 0; i < 10; i++) {
            String key = "key-" + i;
            String value = "Hello Kafka " + i;
            
            ProducerRecord<String, String> record = 
                new ProducerRecord<>("my-topic", key, value);
            
            // 비동기 전송
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception e) {
                    if (e != null) {
                        e.printStackTrace();
                    } else {
                        System.out.println("Sent record: " + 
                            "Topic: " + metadata.topic() + ", " +
                            "Partition: " + metadata.partition() + ", " +
                            "Offset: " + metadata.offset()
                        );
                    }
                }
            });
        }
        
        // 자원 해제
        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

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        // 컨슈머 속성 설정
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-consumer-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("my-topic"));
        
        // 메시지 폴링 및 처리
        try {
            while (true) {
                ConsumerRecords<String, String> records = 
                    consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(
                        "Received record: " +
                        "Topic: " + record.topic() + ", " +
                        "Partition: " + record.partition() + ", " +
                        "Offset: " + record.offset() + ", " +
                        "Key: " + record.key() + ", " +
                        "Value: " + record.value()
                    );
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 자원 해제
            consumer.close();
        }
    }
}

1

pip install kafka-python

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

from kafka import KafkaProducer
import json

# 프로듀서 생성
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# 메시지 전송
for i in range(10):
    data = {'number': i, 'message': f'Hello Kafka {i}'}
    future = producer.send('my-topic', value=data)
    # 결과 확인
    try:
        record_metadata = future.get(timeout=10)
        print(f"Sent to {record_metadata.topic} partition {record_metadata.partition} offset {record_metadata.offset}")
    except Exception as e:
        print(f"Error sending message: {e}")

# 모든 메시지가 전송될 때까지 대기
producer.flush()
producer.close()

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

from kafka import KafkaConsumer
import json

# 컨슈머 생성
consumer = KafkaConsumer(
    'my-topic',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    group_id='my-python-consumer-group',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

# 메시지 소비
for message in consumer:
    print(f"Received: {message.topic} partition {message.partition} offset {message.offset}")
    print(f"Key: {message.key}, Value: {message.value}")

1
2
3
4
5
6
7
8
9

{
  "namespace": "example.avro",
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": ["int", "null"]}
  ]
}

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

{
  "name": "mysql-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:mysql://localhost:3306/mydatabase",
    "connection.user": "user",
    "connection.password": "password",
    "topic.prefix": "mysql-",
    "table.whitelist": "users,orders",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "tasks.max": "1"
  }
}

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

{
  "name": "elasticsearch-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "topics": "mysql-users",
    "connection.url": "http://localhost:9200",
    "type.name": "_doc",
    "key.ignore": "true",
    "schema.ignore": "true",
    "tasks.max": "1"
  }
}

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

-- 스트림 생성
CREATE STREAM orders (
    order_id BIGINT,
    customer_id BIGINT,
    product_id BIGINT,
    quantity INT,
    price DECIMAL(10, 2)
) WITH (
    KAFKA_TOPIC='order-events',
    VALUE_FORMAT='JSON'
);

-- 테이블 생성
CREATE TABLE product_sales AS
SELECT
    product_id,
    SUM(quantity * price) AS total_sales
FROM orders
GROUP BY product_id;

-- 쿼리 실행
SELECT * FROM product_sales WHERE total_sales > 1000;

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

# source-connect.properties
bootstrap.servers=kafka1:9092,kafka2:9092
group.id=mirror-group
config.storage.topic=mm2-configs
offset.storage.topic=mm2-offsets
status.storage.topic=mm2-status

# target-connect.properties
bootstrap.servers=kafka3:9092,kafka4:9092
group.id=mirror-group
config.storage.topic=mm2-configs
offset.storage.topic=mm2-offsets
status.storage.topic=mm2-status

# mm2.properties
clusters=source, target
source.bootstrap.servers=kafka1:9092,kafka2:9092
target.bootstrap.servers=kafka3:9092,kafka4:9092
source->target.enabled=true
source->target.topics=.*
tasks.max=10

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

// 요청 토픽에 메시지 전송
String correlationId = UUID.randomUUID().toString();
ProducerRecord<String, String> record = 
    new ProducerRecord<>("request-topic", correlationId, requestData);
producer.send(record);

// 응답 토픽에서 특정 상관 ID의 메시지만 소비
consumer.subscribe(Arrays.asList("response-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        if (record.key().equals(correlationId)) {
            // 응답 처리
            return record.value();
        }
    }
}

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

# broker 설정
listeners=SSL://localhost:9093
ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
ssl.keystore.password=keystore-password
ssl.key.password=key-password
ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
ssl.truststore.password=truststore-password
ssl.client.auth=required

# 클라이언트 설정
security.protocol=SSL
ssl.truststore.location=/var/private/ssl/kafka.client.truststore.jks
ssl.truststore.password=truststore-password
ssl.keystore.location=/var/private/ssl/kafka.client.keystore.jks
ssl.keystore.password=keystore-password
ssl.key.password=key-password

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

# 브로커 설정
listeners=SASL_SSL://localhost:9093
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

# JAAS 구성
listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="admin" \
    password="admin-secret" \
    user_admin="admin-secret" \
    user_user1="user1-secret";

# 클라이언트 설정
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="user1" \
    password="user1-secret";

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

# 토픽에 대한 읽기 권한 부여
bin/kafka-acls.sh --bootstrap-server localhost:9092 \
    --add --allow-principal User:user1 \
    --operation Read \
    --topic my-topic

# 컨슈머 그룹에 대한 권한 부여
bin/kafka-acls.sh --bootstrap-server localhost:9092 \
    --add --allow-principal User:user1 \
    --operation Read \
    --group my-consumer-group

# 프로듀서에 대한 쓰기 권한 부여
bin/kafka-acls.sh --bootstrap-server localhost:9092 \
    --add --allow-principal User:user2 \
    --operation Write \
    --topic my-topic

1
2
3
4

# server.properties
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@localhost:9093,2@localhost:9093,3@localhost:9093