🙊

Spring Cloud Stream

Intro::

spring-cloud-stream을 사용해서 kafka 메시지를 송수신하는 예제입니다.

studies/Spring/spring-cloud-stream-kafka at main · SONGS4RI/studies

Contribute to SONGS4RI/studies development by creating an account on GitHub.

https://github.com/SONGS4RI/studies/tree/main/Spring/spring-cloud-stream-kafka

kafka cluster 구현

version: "3.8"
services:
  controller-1:
    image: apache/kafka:latest
    container_name: controller-1
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093

  controller-2:
    image: apache/kafka:latest
    container_name: controller-2
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093

  controller-3:
    image: apache/kafka:latest
    container_name: controller-3
    environment:
      KAFKA_NODE_ID: 3
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093

  broker-1:
    image: apache/kafka:latest
    container_name: broker-1
    ports:
      - "29092:9092"
    environment:
      KAFKA_NODE_ID: 4
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: 'PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-1:19092,PLAINTEXT_HOST://localhost:29092'
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
    depends_on:
      - controller-1
      - controller-2
      - controller-3

  broker-2:
    image: apache/kafka:latest
    container_name: broker-2
    ports:
      - "39092:9092"
    environment:
      KAFKA_NODE_ID: 5
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: 'PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-2:19092,PLAINTEXT_HOST://localhost:39092'
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
    depends_on:
      - controller-1
      - controller-2
      - controller-3

  broker-3:
    image: apache/kafka:latest
    container_name: broker-3
    ports:
      - "49092:9092"
    environment:
      KAFKA_NODE_ID: 6
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: 'PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-3:19092,PLAINTEXT_HOST://localhost:49092'
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:9093,2@controller-2:9093,3@controller-3:9093
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
    depends_on:
      - controller-1
      - controller-2
      - controller-3

  kafka-init:
    image: apache/kafka:latest
    container_name: kafka-init
    depends_on:
      - broker-1
      - broker-2
      - broker-3
    volumes:
      - ./scripts:/opt/kafka/scripts
    entrypoint: ["sh", "/opt/kafka/scripts/init-kafka.sh"]
YAML
복사

#!/bin/bash

sleep 10

# Kafka 토픽 생성
/opt/kafka/bin/kafka-topics.sh --bootstrap-server broker-1:19092,broker-2:19092,broker-3:19092 \
  --create --topic test-topic --partitions 3 --replication-factor 1

/opt/kafka/bin/kafka-topics.sh --bootstrap-server broker-1:19092,broker-2:19092,broker-3:19092 \
  --create --topic test-topic-dlq --partitions 3 --replication-factor 1

echo "init kafka"

Bash
복사

controller

kafka broker의 메타데이터를 관리하는 노드이다.

broker

실질적인 카프카 서버

build.gradle

plugins {
    id 'java'
    id 'org.springframework.boot' version '3.3.6'
    id 'io.spring.dependency-management' version '1.1.6'
}

group = 'com.example'
version = '0.0.1-SNAPSHOT'

java {
    toolchain {
        languageVersion = JavaLanguageVersion.of(17)
    }
}

repositories {
    mavenCentral()
    maven { url 'https://repo.spring.io/milestone' }
}

ext {
    set('springCloudVersion', "2023.0.3") // Spring Cloud 2024.0.0-RC1 는 안되는데 왜그런건지는 모르겠다....
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.cloud:spring-cloud-stream'
    implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'
    implementation 'org.projectlombok:lombok'

    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'org.springframework.cloud:spring-cloud-stream-test-binder'
    testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"
    }
}

tasks.named('test') {
    useJUnitPlatform()
}
Plain Text
복사

왜 인지는 모르겠지만 set('springCloudVersion', "2024.0.0-RC1")과 그에 맞는 스프링 부트 버전을 사용하게 되면 MessageBuilder에서 오류가 나서 버전을 한 단계 내렸더니 문제가 해결되었다.

application.yml

spring:
  cloud:
    function:
      definition: input;dlq # 하나 이상의 함수가 정의된 경우, Spring Cloud Stream이 실행할 함수를 선택하기 위해 필요합니다.
    stream:
      bindings:
        input-in-0:
          destination: test-topic
          content-type: application/json
          group: test-group
          consumer:
            maxAttempts: 3 # 최대 3번 재시도
            back-off-initial-interval: 1000 # 1초 간격으로 재시도
            concurrency: 3 # 3개의 컨슈머로 병렬 처리
        dlq-in-0:
          destination: test-topic-dlq
          content-type: application/json
          group: test-dlq-group
      kafka:
        bindings:
          input-in-0:
            consumer:
              enable-dlq: true # DLT 활성화
              dlq-name: test-topic-dlq # DLT 토픽 이름
              autoCommitOnError: false # 오류 발생 시 자동 커밋 방지
        binder:
          brokers: localhost:29092,localhost:39092,localhost:49092
          auto-create-topics: false
YAML
복사

•

바인더는 “애플리케이션과 메시지 브로커를 잇는 드라이버”이며, 실제 구현체 설정을 담당합니다.

•

바인딩은 “애플리케이션의 논리 채널(또는 함수)과 외부 목적지를 연결”하는 설정으로, 어느 채널이 어떤 토픽·큐로 메시지를 주고받을지 정의합니다.

특이사항

•

spring.cloud.function

◦

바인딩 함수를 정의해두지 않으면 바인딩이 제대로 안될수 있습니다.

•

spring.cloud.stream.bindings

◦

input-in-0 이면 Consumer<Message<T>>  input() 과 같은 함수가 바인딩 된다.

◦

output-out-0이면 Supplier<Message<T>> output()과 같은 함수가 바인딩된다.

•

spring.cloud.stream.kafka

◦

카프카 세부 설정부분입니다.

Consumer

import com.example.springcloudstreamkafka.share.domain.Product;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;

import java.util.function.Consumer;

public class MessageConsumer {
    private final ProductMessageQueue productMessageQueue;

    public MessageConsumer(ProductMessageQueue productMessageQueue) {
        this.productMessageQueue = productMessageQueue;
    }

    @Bean
    public Consumer<Message<Product>> input() throws RuntimeException {
        return message -> {
            boolean added = productMessageQueue.buffer.offer(message);
            if (!added) {
                throw new RuntimeException("Queue is full, message discarded: " + message);
            }

            System.out.println("Message added to queue: " + message.getPayload());
        };
    }
}
Java
복사

•

org.springframework.messaging.Message를 사용하여 추상화의 장점을 살리기 때문에 실제로 카프카 메시지가 들어오던 RabbitMQ 메시지가 들어오던 처리가 가능하다.

•

applicatoin.yml에 설정해 놓은 bindings와 binder의 설정을 통해 지정한 토픽에 메시지가 들어오면 비동기적으로 소비하게 됩니다. 이때 람다함수가 해당 토픽을 처리하는 방식이 됩니다.

•

dlq(Dead Letter Queue) 를 설정해두면 자동으로 실패 메시지가 지정된 토픽으로 송신됩니다.

Produce

import com.example.springcloudstreamkafka.share.infra.eventbus.ProductMessageQueue;
import com.example.springcloudstreamkafka.share.domain.Product;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Service;

@Service
public class MessageService {
    private final StreamBridge streamBridge;
    private final ProductMessageQueue productMessageQueue;

    public MessageService(StreamBridge streamBridge, ProductMessageQueue productMessageQueue) {
        this.streamBridge = streamBridge;
        this.productMessageQueue = productMessageQueue;
    }

    public void sendMessage(Product product) {
        boolean sent = streamBridge.send("test-topic", product);
        if (sent) {
            System.out.println("Message sent successfully to test-topic" + " : " + product);
        } else {
            System.out.println("Message sending failed to test-topic" + " : " + product);
        }
    }
}
Java
복사

•

StreamBridge 를 사용해서 메시지를 전송합니다.

◦

이때 토픽이 존재하지 않아도 메시지가 전송되어지고, 설정에 따라 토픽이 생성될 수도 있으니 주의해야합니다.

Function과 Consumer, Supplier의 차이

특성	Function	Consumer	Supplier
입력	있음	있음	없음
출력	있음	없음	있음
사용 예제	데이터 변환 및 가공	메시지 소비 (로그, DB 저장 등)	데이터 생성 (이벤트 발생 등)
바인딩 요구 사항	입력과 출력을 모두 정의 가능	입력 바인딩만 필요	출력 바인딩만 필요

References::

Spring Cloud Stream 정리

데이터 중심 애플리케이션을 단독으로 빌드하고, 테스트하고 배포할 수 있다.메세지 구성 요소에 모던한 마이크로 아키텍처 패턴을 적용할 수 있다.이벤트 중심적 사고에서 애플리케이션간의 책임을 분리할 수 있다. 각 서버의 노드는 이벤트의 원본 발행자를 신경쓰지 않고 이벤트를

https://velog.io/@zenon8485/Spring-Cloud-Stream-%EC%A0%95%EB%A6%AC