AWS SQS With Spring Cloud

• 이번에는 스프링 클라우드 패키지를 이용하여 AWS SQS를 이용해보자.

필수 설정하기

• Spring Cloud 2.3을 이용하여 SQS를 사용할 것이므로, 의존성 라이브러리를 다음과 같이 추가하자.

pom.xml

• 의존성 관리 설정 

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>io.awspring.cloud</groupId>
            <artifactId>spring-cloud-aws-dependencies</artifactId>
            <version>2.3.0</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

• 의존성 추가

<dependencies>
    <dependency>
        <groupId>io.awspring.cloud</groupId>
        <artifactId>spring-cloud-starter-aws-messaging</artifactId>
    </dependency>
</dependencies>

• 관련된 정보는 다음을 참조하자. Spring Cloud AWS 2.3 is now available

Message Sender

• Spring Cloud 에서는 SQS를 지원해 주기 위해서 다음 2가지 방법을 제공한다.
  • QueueMessagingTemplate 는 메시지를 보내는 편리한 메소드를 제공하고 있다.
  • QueueMessageChannel 는 지정된 대상에 메시지를 보내는 메소드를 가지고 있으며, 이 전송 대상은 SQS API 에 대해서 문자열을 사용하여 대상을 지정한다.

QueueMessagingTemplate 로 서비스 구현하기

• QueueMessagingTemplate 는 편리한 많은 메소드를 제공한다.
• 아래 코드는 Template를 이용하여 메시지 전송을 수행하는 방법을 구현하고 있다.
• com.schooldevops.sqs.jms.MessageSenderWithTemplate.java 에 다음과 같이 작성하자.

package com.schooldevops.sqs.jms;

import io.awspring.cloud.messaging.core.QueueMessagingTemplate;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class MessageSenderWithTemplate {

    private static final String QUEUE = "my-todo2";

    @Autowired
    private QueueMessagingTemplate messagingTemplate;

    public void send(String messagePayload) {

        Message msg = MessageBuilder.withPayload(messagePayload)
                .setHeader("sender", "app1")
                .build();
        messagingTemplate.convertAndSend(QUEUE, msg);
        log.info("message sent");
    }

    public void sendToFifoQueue(String messagePayload, String messageGroupID, String messageDedupID) {

        Message msg = MessageBuilder.withPayload(messagePayload)
                .setHeader("message-group-id", messageGroupID)
                .setHeader("message-deduplication-id", messageDedupID)
                .build();
        messagingTemplate.convertAndSend(QUEUE, msg);
        log.info("message sent");
    }
}

• 템플릿을 이용하기 위해서 아래와 같이 Autowire 하였다.

    @Autowired
    private QueueMessagingTemplate messagingTemplate;

• 그리고 전송할 메시지를 MessageBuilder를 이용하여 생성한다.
• 메시지 빌더를 통해서 메시지 헤더를 추가할 수 있고, 메시지 페이로드를 아래와 같이 실어 보낸다 .

        Message msg = MessageBuilder.withPayload(messagePayload)
                .setHeader("sender", "app1")
                .build();

• 위 코드는 헤더로 sender: app1 으로 지정했다.

• withPayload에 실제 전송할 메시지를 담는다.

• 그리고 아래와 같이 Queue이름을 지정하고, 메시지를 전송한다.

        messagingTemplate.convertAndSend(QUEUE, msg);

FiFoQueue로 전송하기

• FifoQueue로 전송하고자 한다면 메시지 헤더에 FifoQueue를 나타내는 값을 실어서 보낼 수 있다.

        Message msg = MessageBuilder.withPayload(messagePayload)
                .setHeader("message-group-id", messageGroupID)
                .setHeader("message-deduplication-id", messageDedupID)
                .build();

• Fifoqueue는 "message-group-id" 로 메시지 그루핑이 필요하다.

• 또한 message-deduplication-id 를 지정하여 중복을 제거한다. 중복을 구분할 내용은 메시지 중복 키를 받아서 전송하고 있다.

• 그리고 메시지 전송은 동일하게 전송한다. 큐 이름과, 메시지를 담아서 보낸다.

QueueMessageChannel 로 서비스 구현하기

• 이번에는 QueueMessageChannel을 이용한 방법으로 메시지 Sender를 만들어 보자.
• com.schooldevops.sqs.jms.MessageSender.java 파일을 생성하고 아래와 같이 작성하자.

package com.schooldevops.sqs.jms;

import com.amazonaws.services.sqs.AmazonSQSAsync;
import io.awspring.cloud.messaging.core.QueueMessageChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class MessageSender {
    private static final Logger logger = LoggerFactory.getLogger(MessageSender.class);

    private static final String QUEUE_NAME = "https://sqs.ap-northeast-2.amazonaws.com/103382364946/my-todo2";

    @Autowired
    private final AmazonSQSAsync amazonSqs;

    @Autowired
    public MessageSender(final AmazonSQSAsync amazonSQSAsync) {
        this.amazonSqs = amazonSQSAsync;
    }

    public boolean send(String messagePayload) {
        MessageChannel messageChannel = new QueueMessageChannel(amazonSqs, QUEUE_NAME);

        Message msg = MessageBuilder.withPayload(messagePayload)
                .setHeader("sender", "app1")
                .build();

        long waitTimeoutMillis = 5000;
        boolean sentStatus = messageChannel.send(msg,waitTimeoutMillis);
        logger.info("message sent");
        return sentStatus;
    }
}

• SQS 라이브러리 로드하기

    private static final String QUEUE_NAME = "https://sqs.ap-northeast-2.amazonaws.com/103382364946/my-todo2";

    @Autowired
    private final AmazonSQSAsync amazonSqs;

• QueueMessageChannel 을 위해서 큐 이름을 SQS 의 url 을 복사해서 사용한다. 이를 SQS API 가 해석해서 사용하게 된다.
• AmazonSQSAsync 를 Autowire 하여 이를 이용한다.

메시지 채널 생성하기

• 메시지 채널은 다음과 같이 생성한다.

MessageChannel messageChannel = new QueueMessageChannel(amazonSqs, QUEUE_NAME);

• 위 내용과 같이 QueueMessageChannel을 생성할때 amazonSQS와 Queue_Name 을 이용하여 생성하고 있다.

메시지 생성하기

• 메시지 생성은 동일하게 MessageBuilder를 이용하여 그대로 생성한다.

Message msg = MessageBuilder.withPayload(messagePayload)
                .setHeader("sender", "app1")
                .build();

메시지 전송하기

• 메시지 전송은 messageChannel의 메소드를 이용하여 전송한다.

        long waitTimeoutMillis = 5000;
        boolean sentStatus = messageChannel.send(msg, waitTimeoutMillis);

• 위 내용과 같이 msg를 전송하며, 전송 타임아웃인 waitTimeMillis동안 대기하도록 설정한다.

Message Receiver

• 이제는 메시지 리시버를 생성해 보자.
• 메시지 리시버는 @SqsListener 를 이용하여 특정 Queue에 대해서 메시지를 수신하도록 리슨한다.
• com.schooldevops.sqs.jms.MessageReceiver.java 파일을 생성하고 다음과 같이 소스를 작성하자.

package com.schooldevops.sqs.jms;


import io.awspring.cloud.messaging.listener.SqsMessageDeletionPolicy;
import io.awspring.cloud.messaging.listener.annotation.SqsListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class MessageReceiver {
    @SqsListener(value = "my-todo2", deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS )
    public void receiveStringMessage(String message,
                                     @Header("SenderId") String senderId) {
        log.info("message received {} {}",senderId,message);
    }
}

• 위 코드와 같이 메소드 단위의 어노테이션인 @SqsListener 를 추가해 주는 것으로 메시지를 수신할 수 있다.
  • value: 큐 이름
  • deletionPolicy: 메시지를 삭제 정책을 지정한다. 위 내용 (SqsMessageDeletionPolicy.ON_SUCCESS)은 성공적으로 처리가 끝나면 메시지를 삭제하게 된다.

properties 설정하기

• 이제 Cloud가 BootUp 될때 이용할 프로퍼티를 아래와 같이 작성하자.
• aplication.properties 파일에 내용을 추가한다.

cloud.aws.credentials.profile-name=terraform_user
cloud.aws.region.auto=false
cloud.aws.region.static=ap-northeast-2

• 프로파일 이름을 지정하고, 리젼 설정을 해 주었다.

테스트하기

컨트롤러 생성하기

• com.schooldevops.sqs.jms.controllers.MessageController.java 에 코드를 작성하자.

package com.schooldevops.sqs.jms.controllers;

import com.schooldevops.sqs.jms.MessageSender;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
public class MessageController {

    @Autowired
    MessageSender messageSender;

    @PostMapping("/message/{msg}")
    public String sendMessage(@PathVariable("msg") String message) {
        messageSender.send(message);
        log.info("Sent Message : {}", message);
        return message;
    }
}

• 메시지 센더를 등록하고 전송하는 간단한 내용이다.

결과

curl -X POST http://localhost:8080/message/Hello-THUB

2021-08-12 18:09:01.235  INFO 30312 --- [           main] c.s.sqs.jms.MessageSenderWithTemplate    : message sent
2021-08-12 18:09:01.269  INFO 30312 --- [enerContainer-2] c.schooldevops.sqs.jms.MessageReceiver   : message received AIDARQEQRWMJJSR7Q72NQ {"payload":"Hello queue","headers":{"id":"f1db446c-da77-9f36-2ec6-fdbb2d1d445e","sender":"app1","timestamp":1628759340901}}

• 위 내용과 같이 message sent 로 메시지가 전송 되었고
• message receive 값이 출력된다. 이는 @SqsListener 에 의해서 수신된 값이다.
  • payload: 우리가 전송한 메시지 페이로드가 담겨 있다.
  • headers: 전송한 헤더 정보가 존재한다. 메시지 id, 그리고 sender (우리가 소스에서 지정해 주었다.) 가 app1 인것을 알 수 있다.

결론

• 지금까지 본것과 같이 Spring-Cloud 를 이용하면, JMS를 이용한것과는 다르게 몇가지 설정 및 라이브러리로 편리하게 미시지를 전송, 수신 할 수 있음을 확인할 수 있다.
• 개발 환경에 따라 JMS를 이용할지, Spring-Cloud-AWS 를 이용할지 선택하면 될듯 하다.
• 소스코드는 Github 을 참조하자.