Amazon SNS(Simple Notification Service)
게시자(Publishers, 생산자)에서 구독자(Subscribers, 소비자)에게 메시지 전달(전송)을 조정, 관리하는 관리형 서비스
게시자 - 주제에 대한 메시지를 생산, 발송함으로써 구독자와 비동시적으로 통신하는 논리적 액세스 및 커뮤니케이션 채널
구독자 - 주제를 구독하는 경우 지원되는 프로토콜 중 하나를 통해 메시지 또는 알림을 소비하거나 수신
웹 서버, 이메일 주소, Amazon SQS 대기열, Lambda 함수
주제(topic)
-
통신 채널 역할을 하는 논리적 액세스 포인트
-
주제를 사용해 여러 엔드포인트(예: Lambda, SQS, HTTP/S, 이메일, … 등)를 그룹화할 수 있음
구독(subscribe)
-
주제에 개시된 메시지를 수신할 수 있도록 주제에 대한 엔드 포인트를 등록하는 것
게시(publish)
-
주제에 메시지를 등록
사용 예)
-
Fadeout
-
경고 : 사전 정의된 임계값에 의해 트리거되는 알림
-
사용자에게 문자 메시지(SMS) 또는 이메일 등으로 알림
-
모바일 푸시 알림
Creating and Subscribing to AWS SNS Topics
주제(topic) 생성
FIFO - 순서가 중요한 경우
표준 - 순서가 중요하지 않은 경우
이메일 구독(subscribe) 생성
SMS 구독 생성
람다 함수 생성
AWS Lambda 함수 핸들러(Python)
https://docs.aws.amazon.com/ko_kr/lambda/latest/dg/python-handler.html
-
첫 번째 인수는 이벤트 객체입니다. 이벤트는 Lambda 함수가 처리할 데이터가 포함된 JSON 형식 문서입니다. Lambda 런타임은 이벤트를 객체로 변환한 후 함수 코드에 전달합니다. 이 객체는 일반적으로 Python dict 유형입니다. 또한 list, str, int, float 또는 NoneType 유형이 될 수 있습니다.
이벤트 객체에는 호출 서비스의 정보가 포함됩니다. 함수를 호출할 때, 이벤트의 구조와 내용을 결정합니다. AWS 서비스는 함수를 호출할 때 이벤트 구조를 정의합니다. AWS 서비스의 이벤트에 대한 자세한 내용은 다른 서비스와 함께 AWS Lambda 사용 단원을 참조하십시오. -
두 번째 인수는 컨텍스트 객체입니다. 컨텍스트 객체는 런타임에 Lambda에 의해 함수로 전달됩니다. 이 객체는 호출, 함수 및 런타임 환경에 관한 정보를 제공하는 메서드 및 속성들을 제공합니다.
SNS topic notification을 통해서 람다 함수가 호출되는 경우, event 인자를 통해서 전달되는 값의 예
{
"Records": [
{
"EventSource": "aws:sns",
"EventVersion": "1.0",
"EventSubscriptionArn": "arn:aws:sns:us-east-1:{{{accountId}}}:ExampleTopic",
"Sns": {
"Type": "Notification",
"MessageId": "95df01b4-ee98-5cb9-9903-4c221d41eb5e",
"TopicArn": "arn:aws:sns:us-east-1:123456789012:ExampleTopic",
"Subject": "example subject",
"Message": "example message", ⇐ event['Records'][0]['Sns']['Message']
"Timestamp": "1970-01-01T00:00:00.000Z",
"SignatureVersion": "1",
"Signature": "EXAMPLE",
"SigningCertUrl": "EXAMPLE",
"UnsubscribeUrl": "EXAMPLE",
"MessageAttributes": {
"Test": {
"Type": "String",
"Value": "TestString"
},
"TestBinary": {
"Type": "Binary",
"Value": "TestBinary"
}
}
}
}
]
}
람다 함수 구독 생성
메시지 게시(publish)
메시지 확인
이메일 구독
SMS 구독
Lambda 함수 구독 ⇒ 함수 실행 여부를 로그를 통해서 확인
Configuring SNS Push Notifications on S3 Bucket Events Inside of the AWS Console
실습1. Configuring SNS Push Notifications on S3 Bucket Events Inside of the AWS Console
### S3 버킷 생성
### SNS 주제(topic) 생성
### S3 버킷에 이벤트 알림을 추가
⇒ SNS 게시 권한이 없기 때문에 오류가 발생
### SNS 주제 정책을 수정
### S3 버킷에 이벤트 알림을 추가 (다시 시도)
### 이메일 구독 추가하고 테스트
S3 버킷에 파일을 업로드
이메일 수신 확인
### SMS 구독을 생성하고 테스트
S3 버킷에 파일을 업로드 후 문자 메시지를 확인
실습2. 혹시, 실습1을 완료한 사람은 메시지를 로그로 남기는 람다 함수를 만들어서 구독에 추가해 보세요.
### 람다 함수를 생성
람다를 호출하는 서비스에 따라서 전달되는 event의 구조가 상이하기 때문에 확인 필요
### 람다 함수 구독을 생성
### S3 버킷에 오브젝트를 등록했을 때 람다 함수가 호출되는지를 확인
실습3. 혹시, 실습2도 완료한 사람은 S3 버킷에 오브젝트가 추가되었을 때 추가된 오브젝트 이름을 로그로 남기는 람다 함수를 호출하도록 해 보세요.
### 람다 함수를 생성
업로드된 객체 정보
### S3 버킷에 이벤트 알림을 추가
-
기존 이번트 알림을 삭제하고 진행
### S3 버킷에 오브젝트 등록 후 람다 함수의 실행 로그를 확인
폴더를 업로드한 경우
S3 버킷에 파일이 업로드되었을 때 SMS 문자가 발송되도록 람다 함수를 수정
⇒ 해당 람다 함수가 SNS:Publish 권한이 없어서 오류가 발생
2번 과정 -> 빨간색
3번 과정 -> 초록색
SQS
Queue
-
메시지를 담는 공간
-
리전 별로 생성해야 함
-
HTTP 프로토콜을 이용해서 다른 리전끼리 메세지를 주고 받을 수 있음
-
큐 이름은 모든 리전에서 유일해야 함
-
큐에 담을 수 있는 메시지의 개수는 무제한
-
연속해서 30일 동안 아무런 요청이 발생하지 않으면 AWS가 큐를 삭제
-
같은 리전 안에서의 데이터 전송은 무료
-
다른 리전에 있는 인스턴스와 메시지를 주고 받으면 데이터 요금이 과금
Message
-
XML 또는 JSON 형태
-
최대 256KB
-
유니코드 문자 사용 가능
-
초 단위로 보관 기간 설정이 가능
-
고유한 ID가 부여됨
-
3 ~ 4KB의 작은 메시지도 256KB로 책정 ⇒ 용량이 작은 메시지를 개별적으로 처리하는 것 보다는 메시지를 모아서 배치 API로 처리하는 것이 요금 효율적
Batch API
-
한번에 최대 10개의 메시지(최대 256KB까지)를 동시에 처리
Visibility Timeout
-
메시지를 받은 뒤 특정 시간 동안 다른 곳에서 동일한 메시지를 다시 꺼내 볼 수 있게 또는 없게 하는 기능
-
큐 하나에 여러 서버에 메시지를 받을 때 동일한 메시지를 동시에 처리하는 것을 방지
-
Message Avaliable (Visible) ⇒ 메시지를 꺼내서 볼 수 있는 상태의 메시지 개수
-
Message in Fight (Not Visible) ⇒ 다른 곳에서 메시지를 보고 있어 현재 볼 수 없는 메시지의 개수
Delay Delivery
-
특정 시간 동안 메시지를 받지 못하게 하는 기능
-
지연되는 시간 동안에는 Message in Fight에 포함
Dead Letter Queues
-
일반적으로 메시지를 받고 작업이 완료되면 메시지를 삭제하는데,
-
설정한 회수를 초과하여 메시지를 받았는데 삭제되지 않고 남아 있으면 처리 실패 큐로 보냄
-
일반 큐 하나에 여러 처리 실패 큐를 연결할 수 있으며, 일반 큐와 처리 실패 큐는 같은 리전에 있어야 함
Short Polling
-
메시지 받기 요청을 하면 결과를 바로 받음
-
메시지가 있으면 메시지를 가져오고 없으면 빠져 나옴
-
ReceiveMessage 요청에서 WaitTimeSeconds를 0으로 했을 때
-
Queue 설정의 ReceiveMessageWaitTimeSeconds를 0으로 했을 때
Long Polling
-
메시지가 있으면 바로 가져오고 없으면 올 때까지 기다림
-
1초부터 20초까지 설정이 가능 (기본은 20초)
-
ReceiveMessage 요청에서 WaitTimeSeconds를 0 보다 크면 큐 설정의 ReceiveMessageWaitTimeSeconds 값 보다 우선으로 처리
Working with AWS SQS Standard Queues
첫번째 터미널에서 표준 SQS 큐를 생성
[cloud_user@ip-10-0-1-23 ~]$ cat create_queue.py
import boto3
# Create SQS client
sqs = boto3.client('sqs')
# Create a SQS queue
response = sqs.create_queue(
QueueName='mynewq',
Attributes={
'DelaySeconds': '5',
'MessageRetentionPeriod': '86400' # 24시간
}
)
print(response['QueueUrl'])
[cloud_user@ip-10-0-1-23 ~]$ python3 create_queue.py
https://queue.amazonaws.com/850718585343/mynewq
create_queue()
response = client.create_queue(
QueueName='string',
Attributes={
'string': 'string'
},
tags={
'string': 'string'
}
)
-
DelaySeconds – The length of time, in seconds, for which the delivery of all messages in the queue is delayed. Valid values: An integer from 0 to 900 seconds (15 minutes). Default: 0.
-
MessageRetentionPeriod – The length of time, in seconds, for which Amazon SQS retains a message. Valid values: An integer from 60 seconds (1 minute) to 1,209,600 seconds (14 days). Default: 345,600 (4 days).
생성된 큐 URL을 sqs_url.py 파일에 등록 ⇒ 다른 어플리케이션에서 참조하기 위해서 사용
[cloud_user@ip-10-0-1-23 ~]$ sudo vi sqs_url.py
QUEUE_URL = 'https://queue.amazonaws.com/850718585343/mynewq'
[cloud_user@ip-10-0-1-23 ~]$ cat sqs_url.py
QUEUE_URL = 'https://queue.amazonaws.com/850718585343/mynewq'
새 터미널에서 큐 모니터링
[cloud_user@ip-10-0-1-23 ~]$ python3 queue_status.py
0 ApproximateNumberOfMessages
0 ApproximateNumberOfMessagesNotVisible
0 ApproximateNumberOfMessagesDelayed
0 ApproximateNumberOfMessages
0 ApproximateNumberOfMessagesNotVisible
0 ApproximateNumberOfMessagesDelayed
:
|
import boto3 import time from sqs_url import QUEUE_URL sqs = boto3.client('sqs') i = 0 while i < 100000: i = i + 1 time.sleep(1) response = sqs.get_queue_attributes( QueueUrl=QUEUE_URL, AttributeNames=[ 'ApproximateNumberOfMessages', 'ApproximateNumberOfMessagesNotVisible', 'ApproximateNumberOfMessagesDelayed', ] ) for attribute in response['Attributes']: print( response['Attributes'][attribute] + ' ' + attribute ) print('') print('') print('') print('') |
get_queue_attributes
response = client.get_queue_attributes(
QueueUrl='string',
AttributeNames=[
'All'|'Policy'|'VisibilityTimeout'|'MaximumMessageSize'|'MessageRetentionPeriod'|'ApproximateNumberOfMessages'|'ApproximateNumberOfMessagesNotVisible'|'CreatedTimestamp'|'LastModifiedTimestamp'|'QueueArn'|'ApproximateNumberOfMessagesDelayed'|'DelaySeconds'|'ReceiveMessageWaitTimeSeconds'|'RedrivePolicy'|'FifoQueue'|'ContentBasedDeduplication'|'KmsMasterKeyId'|'KmsDataKeyReusePeriodSeconds'|'DeduplicationScope'|'FifoThroughputLimit',
]
)
-
ApproximateNumberOfMessages – Returns the approximate number of messages available for retrieval from the queue.
-
ApproximateNumberOfMessagesDelayed – Returns the approximate number of messages in the queue that are delayed and not available for reading immediately. This can happen when the queue is configured as a delay queue or when a message has been sent with a delay parameter.
-
ApproximateNumberOfMessagesNotVisible – Returns the approximate number of messages that are in flight. Messages are considered to be in flight if they have been sent to a client but have not yet been deleted or have not yet reached the end of their visibility window.
세번째 터미널에서 큐로 메시지를 전송
slow_producer.py
|
import boto3 import json import time from sqs_url import QUEUE_URL # Create SQS client sqs = boto3.client('sqs') with open('slow_data.json', 'r') as f: data = json.loads(f.read()) for i in data: msg_body = json.dumps(i) response = sqs.send_message( QueueUrl=QUEUE_URL, MessageBody=msg_body, DelaySeconds=10, MessageAttributes={ 'JobType': { 'DataType': 'String', 'StringValue': 'NewDonor' }, 'Producer': { 'DataType': 'String', 'StringValue': 'Slow' } } ) print('Added a message with 10 second delay - SLOW') print(response) time.sleep(1) |
send_message
response = client.send_message(
QueueUrl='string',
MessageBody='string',
DelaySeconds=123,
MessageAttributes={
'string': {
'StringValue': 'string',
'BinaryValue': b'bytes',
'StringListValues': [
'string',
],
'BinaryListValues': [
b'bytes',
],
'DataType': 'string'
}
},
MessageSystemAttributes={
'string': {
'StringValue': 'string',
'BinaryValue': b'bytes',
'StringListValues': [
'string',
],
'BinaryListValues': [
b'bytes',
],
'DataType': 'string'
}
},
MessageDeduplicationId='string',
MessageGroupId='string'
)
fast_producer.py
|
import boto3 import json import time from sqs_url import QUEUE_URL # Create SQS client sqs = boto3.client('sqs') with open('fast_data.json', 'r') as f: data = json.loads(f.read()) for i in data: msg_body = json.dumps(i) response = sqs.send_message( QueueUrl=QUEUE_URL, MessageBody=msg_body, DelaySeconds=2, MessageAttributes={ 'JobType': { 'DataType': 'String', 'StringValue': 'NewDonor' }, 'Producer': { 'DataType': 'String', 'StringValue': 'Fast' } } ) print('Added a message with 1 second delay - FAST') print(response) time.sleep(1) |
네번째 터미널에서 큐로부터 메시지를 빠르게 받아 오는 것을 실행
fast_consumer.py
|
import boto3 import json import time from sqs_url import QUEUE_URL # Create SQS client sqs = boto3.client('sqs') i = 0 while i < 10000: i = i + 1 rec_res = sqs.receive_message( QueueUrl=QUEUE_URL, MessageAttributeNames=[ 'All', ], MaxNumberOfMessages=1, VisibilityTimeout=5, WaitTimeSeconds=10 ) del_res = sqs.delete_message( QueueUrl=QUEUE_URL, ReceiptHandle=rec_res['Messages'][0]['ReceiptHandle'] ) print("RECIEVED MESSAGE (FAST CONSUMER):") print('FROM PRODUCER: ' + rec_res['Messages'][0]['MessageAttributes']['Producer']['StringValue']) print('JOB TYPE: ' + rec_res['Messages'][0]['MessageAttributes']['JobType']['StringValue']) print('BODY: ' + rec_res['Messages'][0]['Body']) print("DELETED MESSAGE (FAST CONSUMER)") print("") time.sleep(2) |
response = client.receive_message(
QueueUrl='string',
AttributeNames=[
'All'|'Policy'|'VisibilityTimeout'|'MaximumMessageSize'|'MessageRetentionPeriod'|'ApproximateNumberOfMessages'|'ApproximateNumberOfMessagesNotVisible'|'CreatedTimestamp'|'LastModifiedTimestamp'|'QueueArn'|'ApproximateNumberOfMessagesDelayed'|'DelaySeconds'|'ReceiveMessageWaitTimeSeconds'|'RedrivePolicy'|'FifoQueue'|'ContentBasedDeduplication'|'KmsMasterKeyId'|'KmsDataKeyReusePeriodSeconds'|'DeduplicationScope'|'FifoThroughputLimit',
],
MessageAttributeNames=[
'string',
],
MaxNumberOfMessages=123,
VisibilityTimeout=123,
WaitTimeSeconds=123,
ReceiveRequestAttemptId='string'
)
다섯번째 터미널에서 큐로부터 메시지를 천천히 받아 오는 것을 실행
slow_consumer.py
|
import boto3 import json import time from sqs_url import QUEUE_URL # Create SQS client sqs = boto3.client('sqs') i = 0 while i < 10000: i = i + 1 rec_res = sqs.receive_message( QueueUrl=QUEUE_URL, MessageAttributeNames=[ 'All', ], MaxNumberOfMessages=1, VisibilityTimeout=20, WaitTimeSeconds=10 ) del_res = sqs.delete_message( QueueUrl=QUEUE_URL, ReceiptHandle=rec_res['Messages'][0]['ReceiptHandle'] ) print("RECIEVED MESSAGE (SLOW CONSUMER):") print('FROM PRODUCER: ' + rec_res['Messages'][0]['MessageAttributes']['Producer']['StringValue']) print('JOB TYPE: ' + rec_res['Messages'][0]['MessageAttributes']['JobType']['StringValue']) print('BODY: ' + rec_res['Messages'][0]['Body']) print("DELETED MESSAGE (SLOW CONSUMER)") print("") time.sleep(8) |
Working with AWS SQS FIFO Queues
https://aws.amazon.com/ko/sqs/features/
AWS 프리티어로 회원가입을 다음 주 월요일 전까지 해 주세요.
AWS 프리티어
https://www.youtube.com/watch?v=0RJI8m1KB0w&list=PLORxAVAC5fUWCd8arUJZKyryRWzxFGIGQ&index=7
AWS 회원가입
https://www.youtube.com/watch?v=t2Vz_CK3_3I&list=PLORxAVAC5fUWCd8arUJZKyryRWzxFGIGQ&index=8
대학생을 위한 AWS Educate 가입 방법 ⇐ 학교 이메일 주소가 유효한 사람
https://www.youtube.com/watch?v=BVWZY43tkCg
(AWS Educate 가입 대학생을 위한) AWS 활용 및 문제 해결 방법 http://awsblogskr.s3-ap-northeast-2.amazonaws.com/aws-educate-for-student.pdf
'CLOUD > AWS' 카테고리의 다른 글
| 3/8 - AWS 7차시 (0) | 2021.03.08 |
|---|---|
| 3/5 - A Cloud Guru를 이용한 AWS 6차시 (0) | 2021.03.05 |
| 3/3 - A Cloud Guru를 이용한 AWS 4차시 (0) | 2021.03.03 |
| 3/2 - A Cloud Guru를 이용한 AWS 3차시 (0) | 2021.03.02 |
| 2/26 - A Cloud Guru를 이용한 AWS 2차시 (0) | 2021.02.26 |