Spring Cloud Stream

해당 기술을 찾아보고 공부하며 실제로 프로젝트에 녹인 이유는 다음과 같다.

처음에는 회사 프로젝트가 MQTT를 쓰는게 많았고, 그것을 spring cloud stream으로 써볼 까 였지만?

제가 잘 못찾았을 수도 있지만 예제나 방식을 찾지 못하고 있던 찰나!

 

기본적인 mqtt 방식에서 모든 기기의 요청자체를 rabbit MQ로 받아서 msa 방식으로 나누어서 버스형태로 구성했고

추후에 트래픽 량이 많아지거나 했을 때 kafka 등의 도입도 생각이 있기에 코드를 직접 변경하는 방법이 아닌

yaml 에서 수정하면 유연하게 작업이되는 spring cloud stream을 도입하였다.

 

특징.

1. Rabbit MQ , Kafka등 과의 통합을 추상화하여 지원

2. 개발자가 메시징 시스템의 세부 구현에 신경 쓰지 않고 쓸 수 있다.

3. 편리한 어노테이션 기반구성 <- 이방식은 decared되어서 쓸 수 없다. 대신 메소드 방식으로 변경

4. 확장성 및 유연성: 메시지 처리 로직을 쉽게 확장하거나 변경 가능, 다양한 메시지 패턴(요청 -응답, 이벤트 스트리밍등)을 지원

 

해당 특징들로 인해서 결국 개발자는 메시지 발행 및 수신에만 집중 가능하고, 특정 메시지 브로커에 종속되지 않아도 된다.

 

아래에서 RabbitMQ 라이브러리를 통한 방식과 spring cloud stream을 활용한 방식 두가지를 알아 볼 예정이다.

 

Rabbit MQ 라이브러리를 통한 방법

- 장점

  - 커스텀을 할 때 매우 상세하게 만들 수 있다.

  - 여러가지의 토픽을 받거나 다양한 것들을 설정 가능하다.

  - spring cloud stream보다 성능면에서 조금 더 좋다.

- 단점

  - 상세하게 RabbitMQ에 대한 config 설정을 해주어야 한다.

  - 유연하지 못하다.

@Configuration
public class RabbitMQConfig {
    @Value("${spring.rabbitmq.queue.name}")
    private String queueName;
    @Value("${spring.rabbitmq.exchange.name}")
    private String exchangeName;

    @Bean
    public Queue queue() {
        return new Queue(queueName, true);
    }

    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(exchangeName);
    }

    @Bean
    public Binding binding10030(Queue queue, TopicExchange topicExchange) {
        return BindingBuilder.bind(queue).to(exchange()).with("parsed.10030.*");
    }

    @Bean
    public Binding binding10031(Queue queue, TopicExchange topicExchange) {
        return BindingBuilder.bind(queue).to(exchange()).with("parsed.10031.*");
    }

    @Bean
    public Binding binding10032(Queue queue, TopicExchange topicExchange) {
        return BindingBuilder.bind(queue).to(exchange()).with("parsed.10032.*");
    }
}

 

@Configuration
public class RabbitMQListenerConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(3);
        return factory;
    }
}

 

기본적으로 래빗 엠큐를 스프링에서 쓰기위해 설정파일 세팅이 필요하다
기본 RabbitMQ 설정파일과 Listener 파일을 설정해주어야 한다.
예를들어 커스텀하게 queue 이름을 다르게 두어서 내가 받고 싶은 주제에 대한 큐를
따로 열어서 관리가 가능하다. 그리고 바인딩 할 녀석들을 다 따로 두어서 관리도 가능하다.

 

@Slf4j
@Component
@RequiredArgsConstructor
public class RabbitMQListener {
    private final MqttPublisher mqttPublisher;
    private final MessageParser messageParser;

    @RabbitListener(queues = "message-transformer")
    public void receivedMessage(String message) {
        // 메시지 처리 로직
    }

}

 

리스터 config 를 해놓았다면 어노테이션을 붙여 RabbitListener를 통해 읽을 수 있습니다.
하나의 어노테이션을 쓰기 위해서 설정클래스 두개를 만들어야 한다.

 

Spring Cloud Stream을 이용하는 방법

일단 우선적으로 종속성 추가를 해줍니다.

MVN Repository 접속 후 Spring cloud stream 을 검색

그 결과 나오는 것들 중

Spring Cloud Stream 과 Spring Cloud Stream Binder Rabbit을 build.gradle에 추가합니다.

 

여기서 spring cloud stream은 별도의 설정파일은 없고 application.yml 에서 설정만 해주면됩니다.

spring:
  cloud:
    function:
      definition: receive
    stream:
      bindings:
        receive-in-0:
          destination: receive
          group: receive-group
      rabbit:
        bindings:
          receive-in-0:
            consumer:
              binding-routing-key: parsed.#
              durable-subscription: true
  rabbitmq:
    host: localhost
    port: 35672
    username: user
    password: password

 

- spring.cloud.function.definition: 클래스에서 쓰일 함수명이 receive라고 설정

- stream.bindings.receive-in-0: rabbitmq에 바인될 때의 변수

- destination : 어떤 exchange의 어떤 그룹에서 가져올지를 나타낸다.

- 해당 변수에 대한 rabbit mq 설정은 consumer이고 routing key는 parsed로 시작되는 모든 키이다.

- durable-subscription은 구독자가 오프라인이었다가 다시 온라인 상태로 되거나 오류가 있어서 서버가 내려갔다가 올라오더라도 구독 상태가 유지되는 형식으로 해당 큐와 그안의 메시지들이 유지가 되는 형식

 

이렇게 yaml 설정만 해주면 이제 받을 준비가 끝난 것이다.

 

@Service
@Slf4j
@RequiredArgsConstructor
public class MessageService {

    @Bean
    public Consumer<String> receive() {
        return data -> {
            // 메시지 처리 로직
        };
    }

}

이런식으로 바로 데이터를 받아올 수 있다.

 

그래서 클래스 단위에서 설정을 하지 않기 때문에 kafka로 바뀌게 된다면 yml에서 설정을 수정하여 바로 클래스 변경없이 사용가능하다.