[Spring Cloud로 구현하는 마이크로서비스] Section. 11 데이터 동기화를 위한 Kafka 활용

Apache Kafka

- 오픈소스 메시지 브로커 프로젝트

- 실시간 데이터 피드를 관리하기 위해 통일된 높은 처리량, 낮은 지연 시간을 지닌 플랫폼 제공

 

Kafka가 나오게 된 배경

확장이 어려운 복잡한 시스템 구조

 

데이터가 많아도 확장이 용이한 시스템

 

Kafka Broker

• 실행된 kafka 서버
• 3대 이상의 브로커 클러스터 구성
• 설정 코디네이터로 'Zookeeper' 연동
  • 메타데이터 (Broker ID, Controller ID 저장)
  • 리더 브로커 문제 발생 시 장애 대응하는 역할

 

누가 메시지를 생성하고 소비하는지 신경쓰지않고 메시지를 교환한다.

- kafka client

JAVA, Python, Go, C++ 등 많은 언어를 지원한다.

 

Kafka 실행해보기

Zookeeper 실행

./bin/zookeeper-server-start.sh ./config/zookeeper.properties

 

Kafka 서버 구동

$KAFKA_HOME/bin/kafka-server-start.sh ./config/server.properties

 

토픽 생성 -> 토믹 목록 확인 -> 토픽 정보 확인

*토픽 생성

$KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic quickstart-events --partitions 1

 

*토픽 목록 확인

$KAFKA_HOME/bin/kafka-topics.sh --list --zookeeper localhost:9092

 

메시지 생산&소비

*메시지 생산 (producer)

$KAFKA_HOME/bin/kafka-console-producer.sh --broker-list localhost:9093 --topic quickstart-events

 

*메시지 소비 (consumer)

$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic quickstart-events --from-beginning

 

producer에서 발행한 메시지가 consumer에서 소비되는 것을 확인할 수 있다.

 

Kafka Connect

• 코드 없이 Configuration으로 데이터를 이동
  • RESTful API를 통해 지원
  • Stream 또는 Batch형태로 데이터 전송 가능
  • 커스텀 Connertor를 통한 다양한 Plugin 제공
  
• 서로 다른 시스템의 데이터를 kafka connect, cluster, sink를 통해 이동가능하다.

 

confluent에서 kafka-connect 설치하여 실행.

 

connect프로퍼티 파일 설정에

plugin.path=$KAFKA_JDBC_HOME/lib 설정하여 jdbc 사용할 수 있도록 설정 (jdbc)

$KAFKA_CONNECT_HOME/share/java/kafka 디렉토리에 mysql-connector-java-version.jar 파일을 복사해둔다. (mysql)

 

Kafka Source Connect 사용

echo '

{

    "name" : "my-source-connect",

    "config" : {

        "connector.class" : "io.confluent.connect.jdbc.JdbcSourceConnector",

        "connection.url":"jdbc:mysql://localhost:3306/mydb",

        "connection.user":"root",

        "connection.password":"test1357",

        "mode": "incrementing",

        "incrementing.column.name" : "id",

        "table.whitelist":"users",

        "topic.prefix" : "my_topic_",

        "tasks.max" : "1"

	}

}

' | curl -X POST -d @- http://localhost:8083/connectors --header "content-Type:application/json"

localhost:8083 (카프카 커넥터)가 제공하는 엔드포인트 API를 활용하여 새로운 source connector를 등록할 수 있다.

커넥트 이름, 커넥터클래스, url정보, 증가컬럼 등이 제공된다. 

whitelist -> 감지 대상 테이블

MySql 데이터 추가

• users 추가 데이터

insert into users(user_id, pwd, name) value('user1', 'test1111', 'User name');

 

users 테이블에서 변경사항이 생기면 최초에는 topic이 생기면서 데이터를 publish한다. (데이터를 토픽에 쌓음)

consumer에서는 이 데이터를 받아볼 수 있다.

 

•  consumer에서 받는 데이터

{
    "schema": {
        "type": "struct",
        "fields": [
            {
                "type": "int32",
                "optional": false,
                "field": "id"
            },
            {
                "type": "string",
                "optional": true,
                "field": "user_id"
            },
            {
                "type": "string",
                "optional": true,
                "field": "pwd"
            },
            {
                "type": "string",
                "optional": true,
                "field": "name"
            },
            {
                "type": "int64",
                "optional": true,
                "name": "org.apache.kafka.connect.data.Timestamp",
                "version": 1,
                "field": "created_at"
            }
        ],
        "optional": false,
        "name": "users"
    },
    "payload": {
        "id": 1,
        "user_id": "user1",
        "pwd": "test1111",
        "name": "User name",
        "created_at": 1646540630000
    }
}

Kafka Sink Connect 사용

토픽에 전달된 데이터를 sink로 가져와서 사용한다.

* 연결할 정보를 설정한다.

* 토픽을 지정해 사용하는 토픽과 같은 스키마와 데이터의 테이블을 자동 생성한다.

{
    "name": "my-sink-connect",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "connection.url": "jdbc:mysql://localhost:3306/mydb",
        "connection.user": "root",
        "connection.password": "test1357",
        "auto.create": "true",
        "auto.evolve": "true",
        "delete.enabled": "false",
        "tasks.max": "1",
        "topics": "my_topic_users"
    }
}

endpoint요청으로 새로 생긴 my_topic_users 테이블

users테이블에 새로운 데이터를 추가해보자.

insert into users(user_id, pwd, name) value('admin1234', 'admin1234', 'SUPER_ADMIN');

 

토픽에 데이터가 보내지고 이를 인지한 sink가 my_topic_users테이블도 업데이트한다.

 

Console로 데이터 발행

콘솔화면

토픽 발행 시에는 sink와 연결된 my_topic_users에만 데이터가 추가되고 users에는 추가되지 않는다.

Source는 user테이블 변경 발생 시 토픽에 ​데이터를 발행하는 역할이기 때문이다.

실제 서비스 데이터 동기화 적용하기

데이터 동기화 Order -> Catalogs

- Orders Service에 요청 된 주문 수량 정보를 Catalogs Service에 반영

- Order서비스에서 kafka Topic으로 메시지 전송 (Producer)

- Catalog서비스에서 kafka Topic으로 메시지 취득 (Consumer)

 

각각의 서비스가  서로 다른 데이터베이스를 가지고 있으므로 주문이 발생하면 상품의 수량정보를 동기화해줘야한다.

Catalog - 카프카 컨슈머 빈 등록

@EnableKafka
@Configuration
public class KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        HashMap<String, Object> properties = new HashMap<>();
        // kafka container host
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1.9093"); // 카프카 서버의 주소
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerGroupId");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // key
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // value - Deserializer로 풀어서 사용
        return new DefaultKafkaConsumerFactory<>(properties);
    }

    /**
     * 토픽의 변경사항이 생겼을 때 해당 데이터 처리
     * @return
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory
                = new ConcurrentKafkaListenerContainerFactory<>();
        kafkaListenerContainerFactory.setConsumerFactory(consumerFactory());
        return kafkaListenerContainerFactory;
    }
}

카프카 서버와 컨슈머그룹 정보 key, value처리 방식을 정의한 ConsumerFactory를 빈에 등록하고 카프카리스너에 이 정보를 포함하여 빈에 등록한다.

 

Catalog - KafkaConsumer

@Service
@Slf4j
public class KafkaConsumer {
    CatalogRepository repository;

    @Autowired
    public KafkaConsumer(CatalogRepository repository) {
        this.repository = repository;
    }

    /**
     * 토픽에 값이 전달이 되면 그값을 가져와서 실행
     * @param kafkaMessage
     */
    @KafkaListener(topics = "example-catalog-topic")
    public void updateQty(String kafkaMessage) {
        log.info("Kafka Message : {}", kafkaMessage);
        Map<Object, Object> map = new HashMap<>();
        ObjectMapper mapper = new ObjectMapper();
        try {
            map = mapper.readValue(kafkaMessage, new TypeReference<Map<Object, Object>>() {});
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }

        CatalogEntity entity = repository.findByProductId((String) map.get("productId")); //id값으로 catalogEntity정보 가져옴
        if (entity != null) {
            // kafka로 전달된 주문 수량 정보를 잔여수량정보에서 빼서 데이터를 동기화 시킴
            entity.setStock(entity.getStock() - (Integer) map.get("qty"));
            repository.save(entity);
        }

    }
}

'example-catalog-topic'이름의 토픽의 데이터를 받아서 수량을 업데이트 해주는 역할을 담당한다.

 

이제 order-service에 주문을 하면 주문한 수량만큼 catalog-service에 잔여 수량이 kafka를 통해 업데이트 되어 데이터 동기화 된다.

 

Multiple Orders Service에서의 데이터 동기화

인스턴스가 여러개 일때 Orders데이터 동기화 문제

인스턴스가 각각의 데이터베이스를 가지기 때문에 주문 데이터가 서로 다른 데이터베이스에 저장될 수 있다.

그렇기 때문에 특정 유저의 주문 조회시 전체 데이터가 조회되지 않을 수 있다 (한 데이터베이스만 조회 시)

 

Kafka를 이용하여 단일 데이터베이스로 동기화

• 주문 정보를 DB가 아닌 Kafka Topic으로 전송
• Kafka Topic에 설정 된 Kafka Sink Connect를 사용해 단일 DB에 저장 (데이터 동기화)

 

토픽에 전달할 메시지 등록

Topic으로 보낼 데이터의 구성

Topic에 보낼 데이터를 KafkaOrderDto객체로 생성하도록 한다.

@Service
@Slf4j
public class OrderProducer {
    private KafkaTemplate<String, String> kafkaTemplate;

    List<Field> fields = Arrays.asList(
            new Field("string", true, "order_id"),
            new Field("string", true, "user_id"),
            new Field("string", true, "product_id"),
            new Field("int32", true, "qty"),
            new Field("int32", true, "unit_price"),
            new Field("int32", true, "total_price"));

    Schema schema = Schema.builder()
            .type("struct")
            .fields(fields)
            .optional(false)
            .name("orders")
            .build();

    @Autowired
    public OrderProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 주문 들어올 시 호출하여 토픽에 발행한다. (데이터 동기화용)
     * @param topic
     * @param orderDto
     * @return
     */
    public OrderDto send(String topic, OrderDto orderDto) {
        Payload payload = Payload.builder()
                .order_id(orderDto.getOrderId())
                .user_id(orderDto.getUserId())
                .product_id(orderDto.getProductId())
                .qty(orderDto.getQty())
                .unit_price(orderDto.getUnitPrice())
                .total_price(orderDto.getTotalPrice())
                .build();

        //스키마와 페이로드로 구성된 카프카오더dto
        KafkaOrderDto kafkaOrderDto = new KafkaOrderDto(schema, payload);

        ObjectMapper mapper = new ObjectMapper();
        String jsonInString = "";
        try {
            jsonInString = mapper.writeValueAsString(kafkaOrderDto);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }

        kafkaTemplate.send(topic, jsonInString);
        log.info("Order Producer sent data from Order Microservice" + orderDto);
        return orderDto;
    }
}

필드와 스키마는 정해져있으므로 미리 만들어서 사용하고 payload는 요청으로 들어온 데이터를 이용하여 kafka에 보낼 객체를 만들어준다. 이후 객체를 json타입으로 변환하여 정해진 topic으로 send()해주면 order서비스 인스턴스에 상관없이 kafka 토픽에 메시지를 발행한다.

 

이 큐에 쌓인 데이터는 설정해둔 order-sink-connect에 의해 하나의 DB에 저장되게 된다.

 

cqrs 패턴?