Node.js 환경에서 SQS 를 사용해봅니다. (1)

Node.js 환경에서 SQS 를 이용하여 데이터를 주고 받아 봅시다.

준비 사항

AWS 계정 (프리티어를 추천드립니다)

IAM 설정

우리가 만들 애플리케이션 상에서 AWS SDK 를 이용하는데, 이를 사용하기 위해서는 AWS IAM(Identity and Access Management) 에서 제공해주는 엑세스 키 및 시크릿 키 값을 필요로 합니다.
먼저 AWS 콘솔에 로그인하여 IAM 서비스로 접근합니다.

우측 상단의 사용자 추가 버튼을 클릭하면 아래와 같은 페이지가 나오는데 이 때 사용자 이름을 꼭! 기억해주세요, 잠시 후에 요긴하게 사용됩니다.

자격 증명 유형은 엑세스 키를 선택해주세요. 다음 페이지로 넘어가시면 아래와 같이 권한 설정 탭이 나오고 기존 정책 직접 연결에서 SQS 를 검색하면 관련 정책이 나오는데, SQSFullAccess 로 설정해줍니다.

나머지는 전부 다음 클릭하여 넘어가주면 제일 마지막에 엑세스 키 와 시크릿 키 가 같이 나오는 화면에서 꼭 이 엑세스 키와 시크릿 키를 복사하여 보관해주세요

SQS 설정

저는 메세지를 선입선출 형식으로 처리하기를 원해서 FIFO 방식으로 사용하겠습니다.
해당 큐의 이름을 입력해주시는데, 이 때 FIFO 큐는 큐의 이름 뒤에 .fifo 를 붙여주셔야 합니다.

구성 탭에서는 메시지 수신 대기 시간을 제외한 나머지는 전부 기본 설정을 사용하였고 수신 대기 시간은 10초를 주었습니다.

전송자, 수신자 전부 지정된 AWS 계정, IAM 사용자 및 역할을 선택해주시고 안에 들어갈 문자열을 arn:aws:iam::{accountId}:user/{iam} 이런 형태로 입력해야 하는데 이 때 accountId 에 해당하는 값은 AWS 콘솔 우측 최상단 탭에 보이는 사용자 이름을 클릭하면 나옵니다.

해당 번호를 기입해주시고, user 에 해당하는 값은 아까 만들어두었던 IAM 의 사용자 이름을 입력해주시면 됩니다. 이걸 완료하면 arn:aws:iam::123456789012:user/sqs 와 같은 문자열이 나오는데 이것을 각 항목에 입력한 다음 대기열 생성을 누르면 AWS 에서의 설정은 끝났습니다!

Producer

자 이제 실제로 어떤 메세지를 발행할 간단한 Producer 를 구현할 서버 애플리케이션을 작성해보겠습니다. Express 위에서 동작시킬 것이고, AWS SDK 를 활용하기 위해 aws-sdk 를 설치해주셔야 합니다.

npm i --save express aws-sdk

그 다음 이제 우리는 aws-sdk 에서 import 하여 사용할 수 있는 SQS.sendMessage 라는 함수를 이용하여 메세지를 발행할 것 입니다.

sendMessage(params: SQS.Types.SendMessageRequest, callback?: (err: AWSError, data: SQS.Types.SendMessageResult) => void): Request<SQS.Types.SendMessageResult, AWSError>;

먼저 sendMessage() 를 살펴보면, 함수가 오버로딩 되어 있어서 params 라는 매개변수를 넣을 수 있는데 여기서 사용해볼만한 값들을 알아보겠습니다.

//QueueUrl 에는 우리가 만든 SQS 대기열의 상세 페이지에 있는 url
QueueUrl: 'YOUR_QUEUE_URL'
//Consumer 가 수신해서 처리할 데이터
MessageBody: JSON.stringify({title: 'hello world!'})
//DelaySeconds 에 전달된 시간 만큼 대기한 후 Consumer 에서 처리 가능, FIFO 큐에서는 사용 불가
DelaySeconds?: Integer
//해당 속성은 AWS SQS Developer Guide 에 설명이 수록되어 있습니다.
//FIFO 큐에서 필수 값인 메세지가 특정 그룹에 속하는 것을 의미하는 값 입니다.
MessageGroupId: 'test'
//FIFO 큐에서 필수 값인 메세지 중복 제거용 ID 값 입니다. 
//이 값은 모든 메세지에서 고유한 값이어야 합니다.
MessageDeduplicationId: 'test01'

자 이제 10개의 메세지를 발행하는 Producer 애플리케이션을 실행해보겠습니다.

import express from 'express';
import AWS from 'aws-sdk'
const server = express()

//아까 발급한 IAM 의 엑세스 키와 시크릿 키, 그리고 리전을 작성해주세요
AWS.config.update({
    accessKeyId: 'YOUR_ACCESS_KEY',
    secretAccessKey: 'YOUR_SECRET_KEY',
    region: 'YOUR_REGION'
})

const sqs = new AWS.SQS()

server.listen(3000, () => {
    console.log('server running!')
  
  for (let i = 0; i < 10; i++) {
      	//메세지를 발송합니다. 
        sqs.sendMessage({
              QueueUrl: 'YOUR_QUEUE_URL',
              MessageBody: JSON.stringify({title: `${i} 번째 메세지입니다.`}),
              MessageGroupId: 'test',
              MessageDeduplicationId: `test_${i}`
          },
          (err, data) => {
		// 성공했다면 err 값은 null 을 반환하고 data 는 아래와 같은 형식을 가집니다.
      	/** 
      	{
		  ResponseMetadata: { RequestId: '14b98452-6756-55b2-971e-df100fa71667' },
		  MD5OfMessageBody: 'bb8c3cbe6de6209e60eadae40c6c1f25',
		  MessageId: '4e373ed2-86c3-430f-8e35-9498eaa965f8',
		  SequenceNumber: '18871013042047471872'
		}
      	*/
              console.log(err, data)
          })
    }

Consumer

간단한 코드로 메세지 큐에 하나의 메세지를 발송하였습니다. 이제 SQS.receiveMessage() 를 이용한 Consumer 를 구현하여 메세지를 받는 서버 애플리케이션을 작업해보겠습니다.

receiveMessage(params: SQS.Types.ReceiveMessageRequest, callback?: (err: AWSError, data: SQS.Types.ReceiveMessageResult) => void): Request<SQS.Types.ReceiveMessageResult, AWSError>;

receiveMessage(callback?: (err: AWSError, data: SQS.Types.ReceiveMessageResult) => void): Request<SQS.Types.ReceiveMessageResult, AWSError>;

먼저 receiveMessage() 를 살펴보면, 오버로딩 되어 있어서 params 라는 매개 변수를 넣을 수 있는데 여기서 사용해볼만한 값들을 알아보겠습니다.

//QueueUrl 에는 우리가 만든 SQS 대기열의 상세 페이지에 있는 url
QueueUrl: 'YOUR_QUEUE_URL'
//SQS 가 리턴해야 하는 QueueAttributeName 값의 배열입니다. 
AttributeNames?: AttributeNameList
//SQS 가 리턴할 메시지 배열의 길이입니다. 유효한 값은 1 ~ 10입니다. 기본값은 1.
MaxNumberOfMessages?: Integer
//receiveMessage() 에 의해 검색된 후 해당 메시지가 다른 검색에서 숨겨지는 시간(초)입니다.
VisibilityTimeout?: Integer
//큐에 메세지가 없는 상태에서 해당 시간(초) 만큼 기다린 후 receiveMessage() 를 종료시킵니다.
WaitTimeSeconds?: Integer
//이 매개 변수는 FIFO 큐에만 적용됩니다. receiveMessage() 의 중복 제거에 사용되는 토큰입니다.
ReceiveRequestAttemptId?: String;

Producer 에서 발행한 10개의 메세지 중 5개의 메세지만을 받아서 출력하는 Consumer 애플리케이션을 실행해보겠습니다.

import express from 'express';
import AWS from 'aws-sdk'
const server = express()

//아까 발급한 IAM 의 엑세스 키와 시크릿 키, 그리고 리전을 작성해주세요
AWS.config.update({
    accessKeyId: 'YOUR_ACCESS_KEY',
    secretAccessKey: 'YOUR_SECRET_KEY',
    region: 'YOUR_REGION'
})

const sqs = new AWS.SQS()

server.listen(3100, () => {
    console.log('server running!')

    sqs.receiveMessage({QueueUrl: 'YOUR_QUEUE_URL', MaxNumberOfMessages: 5}, (err, data) => {
      	// 성공시 err 는 null 이고 데이터는 아래와 같은 형식을 가집니다.
      	/** 
        {
		  ResponseMetadata: { RequestId: 'd64d1289-88b7-533e-8667-49cbe30261e5' },
		  Messages: [
		    {
		      MessageId: '4e373ed2-86c3-430f-8e35-9498eaa965f8',
		      ReceiptHandle: 'string',
		      MD5OfBody: 'bb8c3cbe6de6209e60eadae40c6c1f25',
		      Body: '{"title":"hello world!"}'
		    }
		  ]
		}
        */
        console.log(err, data)
    })
});

마무리

이렇게 메세지를 발행하는 Producer 와 해당 메세지를 수신하는 Consumer 를 간단하게 구현해보았습니다.
프리티어의 경우 메세지 백만 건 까지는 무료이지만 그 이상 넘어가면 요금이 청구됩니다. 이 부분 주의하셔서 연습을 해주시면 되겠습니다.