Spring Cloud Stream

스프링 클라우드 스트림 소개

스프링 클라우드 스트림 프로젝트는 애플리케이션에 메시지 발행자와 소비자를 쉽게 구축할 수 있는 에너테이션 기반 프레임워크.

사용하는 시징 플랫폼의 구현 세부 사항을 추상화한다. 여러 메시지 플랫폼이 스프링 클라우드 스트림과 사용될 수 있으며, 특정 플랫폼을 위한 세부 구현은 애플리케이션 코드와 분리된다. 애플리케이션에서 메시지 발행과 소비 구현은 플랫폼 중립적인 스프링 인터페이스로 수행된다.

스프링 클라우드 스트림 아키텍처

메시징을 이용한 두 비스의 통신 관점에서 스프링 클라우드 스트림 아키텍처를 살펴보며 논의를 시작해 보자. 한 서비스가 메시지 발행자(publisher)가 되고, 다른 서비스는 메시지 소비자(consumer)가 된다. 아래 그림에서 스프링 클라우드 스트림이 어떻게 메시지를 수월하게 전달하는지 확인할 수 있다.

스프링 클라우드에서 메시지를 발행하고 소비하는 데 다음 4개의 컴포넌트가 관련되어 있다.

• 소스

• 채널

• 바인더

• 싱크

8-3 메시지가 발행되고 소비될 때 하부 메시징 플랫폼을 추상화하는 스프링 클라우드 스트림 컴포넌트들을 통과한다.

소스(SOURCE)

서비스가 메시지를 발행할 준비가 되면 소스를 사용해 메시지를 발행한다. 소스는 발행될 메시지를 표현하는 POJO를 전달받는 스프링의 애너테이션 인터페이스다. 소스는 메시지를 받아 직렬화(기본 직렬화 설정은 JSON)하고 메시지를 채널로 발행한다.

채널(CHANNEL)

메시지 생산자와 소비자가 메시지를 발행하거나 소비한 후 메시지를 보관할 큐를 추상화한 것. 채널 이름은 항상 대상 큐의 이름과 관련이 있지만 코드에서는 큐 이름을 직접 사용하지 않고, 채널 이름을 사용한다. 따라서 채널이 읽거나 쓰는 큐를 전환하려면 애플리케이션 코드가 아닌 구성 정보를 변경한다.

바인더(BINDER)

스프링 클라우드 스트림 프레임워크의 일부인 스프링 코드로 특정 메시지 플랫폼과 통신한다. 스프링 클라우드 스트림의 바인더를 사용하면 메시지를 발행하고 소비하기 위해 플랫폼마다 별도의 라이브러리와 API를 제공하지 않고도 메시징을 사용할 수 있다.

싱크(SINK)

스프링 클라우드 스트림에서 서비스는 싱크를 사용해 큐에서 메시지를 받는다. 싱크는 들어오는 메시지를 위해 채널을 수신 대기하고, 메시지를 다시 POJO로 역직렬화한다. 이 과정에서 스프링 서비스의 비즈니스 로직이 메시지를 처리할 수 있다.

간단한 메시지 Publisher와 Consumer 작성

다음 예제에선 조직 서비스가 라이선싱 서비스로 메시지를 전달한다. 라이선싱 서비스가 메시지를 받을 때 하는 일은 로그 메시지를 콘솔에 출력하는 것뿐이다.

이 제에선 하나의 스프링 클라우드 스트림의 소스(메시지 생산자)와 싱크(메시지 소비자)만 사용하므로, 조직 서비스의 소스와 라이선싱 서비스의 싱크 설정을 손쉽게 해 주는 스프링 클라우드 예제를 시작할 것이다.

8-4 조직 이터가 변경되면 카프카에 메시지를 발행한다

1. 조직 서비스의 메시지 생산자 작성

조직 데이터가 추가/수정/삭제될 때마다 조직 서비스가 카프카 토픽(topic)에 메시지를 발행해서 토픽으로 조직 변경 이벤트가 발생했음을 알려 주도록 조직 서비스를 고쳐 보자. 위 그림에서 메시지 생산자를 설명하여, 그림 8-3의 일반적인 스프링 클라우드 스트림 아키텍처를 기반으로 한다.

발행된 메시지 안에 변경 이벤트와 관련된 조직 ID와 발생한 액션 정보 (Add, Update, Delete)가 포함된다.

먼저 할 일은 조직 서비스의 Maven, Gradle에 의존성을 설정해야 한다.

Maven

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>

Gradle

implementation 'org.springframework.cloud:spring-cloud-stream'
implementation 'org.springframework.cloud:spring-cloud-starter-stream-kafka'

메이븐 의존성을 정의했다면, 메시지 브로커와 바인딩하도록 지정해야 한다. 이 작업을 위해 조직 서비스의 부트스트랩 클래스인 Application.java에 @EnableBinding 에너테이션을 추가한다.

package com.thoughtmechanix.organization;
import com.thoughtmechanix.organization.utils.UserContextFilter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.annotation.Bean;
import javax.servlet.Filter;
@SpringBootApplication
@EnableEurekaClient
@EnableCircuitBreaker
@EnableBinding(Source.class) ----@EnableBinding 애너테이션은 스프링 클라우드 스트림에 애플리케이션을 메시지 브로커로 바인딩하라고 알린다.
public class Application {
    @Bean
    public Filter userContextFilter() {
        UserContextFilter userContextFilter = new UserContextFilter();
            return userContextFilter;
    }
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}
 

@EnableBinding 애너테이션은 해당 서비스를 메시지 브로커에 바인딩하도록 스프링 클라우드 스트림을 설정한다. @EnableBinding 에너테이션에서 Source.class를 사용하면 해당 서비스가 Source 클래스에 정의된 채널들을 이용해 메시지 브로커와 통신하게 된다. 채널은 메시지 큐 위에 있다는 것을 상기하자. 스프링 클라우드 스트림은 메시지 브로커와 통신할 수 있는 기본 채널이 있다.

아직 조직 서비스가 어떤 메시지 브로커를 바인딩할지 스프링 클라우드 스트림에 지정하지 않았다. 이제 메시지를 발행할 코드를 구현할 준비가 되었다.

package com.thoughtmechanix.organization.events.source;

// 간결한 코드를 위해 import 삭제

@Component
public class SimpleSourceBean {
    private Source source;

    private static final Logger logger =
      LoggerFactory.getLogger(SimpleSourceBean.class);

    @Autowired
    public SimpleSourceBean(Source source) { ----스프링 클라우드 스트림은 서비스가 사용할 소스 인터페이스 구현을 주입한다.
        this.source = source;
    }

    public void publishOrgChange(String action, String orgId) {
        logger.debug("Sending Kafka message {}
            for Organization Id: {}",
            action, orgId);
        OrganizationChangeModel change = new OrganizationChangeModel(
            OrganizationChangeModel.class.getTypeName(),
            action,
            orgId,
            UserContext.getCorrelationId()); ----발행될 메시지는 자바 POJO다.

        source
           .output()
           .send(
              MessageBuilder
                .withPayload(change)          .build()); ----메시지를 보낼 준비가 되면 Source 클래스에서 정의된 채널에서 send() 메서드를 사용한다.
    }
}

다음 코드에서 스프링 클라우드의 Source 클래스를 코드에 주입한다. 특정 메시지 토픽에 대한 모든 통신은 채널이라는 스프링 클라우드 스트림 구조로 발생한다는 것을 기억하자. 채널은 자바 인터페이스로 표현되며, 이 코드에서 Source 인터페이스를 사용한다. Source 인터페이스는 스프링 클라우드에서 정의한 인터페이스로 output()이라는 단일 메서드를 노출한다. Source 인터페이스는 서비스가 단일 채널에만 발행해야 할 때 사용하기에편리한 인터페이스이며, output()메서드는 MessageChannel클래스 타입을 반환한다. MessageChannel은 메시지 브로커에 메시지를 보내는 방법을 정의하며, 이 장 뒷부분에서 사용자 정의 인터페이스를 사용해 여러 메시징 채널을노출하는 방법을 보여 줄 것이다.

실제 메시지 발행은 publishOrgChange() 메서드에서 이루어진다. 이 메서드는 OrganizationChange Model이라는 자바 POJO를 만든다. OrganizationChangeModel 클래스는 다음 세 가지 데이터 요소를 담는 POJO 클래스이며, 이 장에서 별도로 코드는 보여 주지 않는다.

• 액션(action) : 이벤트를 발생시킨 액션이다. 메시지에 액션을 포함시키면 메시지 소비자가 이벤트를 처리하는 데 더 많은 컨텍스트를 제공할 수 있다.

• 조직 ID(organization ID) : 이벤트와 연관된 조직ID다.

• 상관관계 ID(correlation ID) : 이벤트를 발생시킨 서비스 호출에 대한 상관관계 ID다. 상관관계 ID는 서비스들을 경유하는 메시지 흐름을 추적하고 디버깅하는 데 도움이 많이 되므로 항상 이벤트에 포함시켜야 한다.

메시지를 발행할 준비가 되면 source.output() 메서드에서 반환되는 MessageChannel 클래스의 send()메서드를 사용한다.

source.output().send(MessageBuilder.withPayload(change).build());

send() 메서드는 스프링 Messag 클래스를 매개변수로 받는다. MessageBuilder라는 스프링 헬퍼 클래스를 사용해 OrganizationChangeModel 클래스 정보를 전달받아 스프링 Message 클래스로 변환한다.

이것이 메시지를 보내는 데 필요한 코드 전부다. 하지만 실제 메시지 브로커뿐만 아니라 특정 메시지 큐에서도 조직 서비스를 바인딩하는 방법을 다루지 않아 아직까지는 이 모든 것이 다소 마술처럼 느껴질 것이다. 실제로 이 모든 일은 구성 설정으로 이루어진다. 이전 코드에서 서비스의 스프링 클라우드 스트림의 Source가 카프카 메시지 브로커와 카프카의 메시지 토픽에 매핑되는 구성을 보여 준다. 이 구성 정보는 서비스의 application.yml 파일이나 스프링 클라우드 컨피그에서 해당 환경에 맞게 설정할 수 있다.

spring:
   application:
      name: organizationservice

   // 간결한 코드를 위해 삭제

      stream: ----stream.bindings는 서비스가 스프링 클라우드 스트림의 메시지 브로커에 발행하려는 구성의 시작점이다.
         bindings:  
            output: ----output은 채널 이름이며, 코드 8-2에서 본 Source.output() 채널에 매핑된다.
               destination: orgChangeTopic ----orgChangeTopic은 메시지를 넣을 메시지 큐(또는 토픽) 이름이다.
               content-type: application/json ----content-type은 스프링 클라우드 스트림에 송수신할 메시지 타입의 정보를 제공한다(이 경우는 JSON이다).
            kafka: ----stream.bindings.kafka 프로퍼티는 해당 서비스가 메시지 버스로 카프카를 사용할 것이라고 스프링에 전달한다(대안으로 RabbitMQ를 사용할 수 있다).
               binder:                  
                  zkNodes: localhost ---- zkNodes와 brokers 프로퍼티는 스프링 클라우드 스트림에 카프카와 주키퍼의 네트워크 위치를 전달한다.
                  brokers: localhost

spring.stream.bindings.output 구성 프로퍼티는 source.output()채널을 통신하려는 메시지 브로커의 orgChangeTopic에 매핑한다. 이 프로퍼티도 topic에 전달하는 메시지를 json으로 직렬화하도록 스프링 클라우드 스트림을 설정한다. 스프링 클라우드 스트림은 JSON 및 XML, 아파치 재단의 아브로 포맷을 포함한 다양한 포맷으로 메시지를 직렬화할 수 있다.

spring.stream.bindings.kafka 구성 프로퍼티는 스프링 클라우드 스트림이 서비스를 카프카에 바인딩하도록 설정한다. 하위 프로퍼티는 카프카 메시지 브로커 및 카프카와 함께 실행되는 아파치 주키퍼 서버의 네트워크 주소를 스프링 클라우드 스트림에 설정한다.

이제 스프링 클라우드 스트림으로 메시지를 발행하는 코드와 카프카를 메시지 브로커로 사용하도록 구성했으니 조직 서비스에서는 실제로 어디서 메시지가 발행되는지 살펴보자. 이 작업은 OrganizationService.java 클래스에서 이루어진다.

package com.thoughtmechanix.organization.services;

// 간결한 코드를 위해 import 삭제

@Service
public class OrganizationService {
    @Autowired    private OrganizationRepository orgRepository;

    @Autowired ----SimpleSourceBean을 OrganizationService에 주입하기 위해 스프링 자동 연결을 사용한다.
    SimpleSourceBean simpleSourceBean;

    // 간결한 코드를 위해 나머지 삭제

    public void saveOrg(Organization org) {
        org.setId(UUID.randomUUID().toString());

        orgRepository.save(org);
        simpleSourceBean.publishOrgChange("SAVE", org.getId()); ----조직 데이터를 변경하는 메서드 모두 simpleSourceBean.publishOrgChange()를 호출한다.
    }
}

Note 어떤 데이터를 메시지에 추가해야 할까?

메시징 개발을 시작한 팀에서 많이 받는 질문 중 하나는 “정확하게 얼마나 많은 데이터를 메시지에 넣어야 하는가” 하는 것이다. 이에 대해 필자는 “애플리케이션마다 다르다.”라고 답한다. 알겠지만 이 장 모든 예제에서는 변경된 조직 레코드의 조직 ID만 반환한다. 필자는 메시지에 변경된 데이터 복사본은 절대 넣지 않는다. 예제(와 전화 통신 분야에서 필자가 겪은 많은 문제)에서 실행되는 비즈니스 로직은 데이터 변화에 민감하다. 필자는 시스템 이벤트에 기반을 둔 메시지를 사용해 다른 서비스에는 데이터가 변경된 사실만 전파하고, 그 서비스들이 항상 마스터(데이터를 소유한 서비스)로 다시 가서 새로운 데이터 사본을 가져오게 했다. 이 방식은 실행 시간 면에서 더 많은 비용이 들지만 항상 최신 데이터 복사본으로 작업하는 것을 보장한다. 작업 중인 데이터를 원본 시스템에서 읽어 온 직후에도 변경될 가능성은 여전히 존재하지만, 큐에서 받은 정보를 무턱대고 사용하는 것보다 가능성이 훨씬 낮다.

얼마나 많은 데이터를 전달할지 신중히 생각해야 한다. 여러분은 조만간 전달한 데이터가 예전(stale) 데이터인 상황에 처할 것이다. 어떤 문제로 메시지 큐에 너무 오래 보관되거나 데이터를 포함한 이전 메시지가 실패해서 일관성 없는 메시지가 전달되기 때문에(애플리케이션이 원래 데이터 스토어에 있는 데이터의 실제 상태보다 메시지 상태에 더 의존하기 때문에) 전달된 데이터는 최신이 아닐 수 있다. 메시지의 상태를 전달하려면 메시지에 날짜-시간 스탬프 또는 버전 번호도 포함시켜 데이터를 소비하는 서비스가 전달된 데이터를 검사하고 이미 가진 데이터 복사본보다 이전 것이 아닌지 확인할 수 있어야 한다(데이터는 비순차적으로 검색될 수 있음을 기억하자).