growing hard days.

AWS LambdaのSQSへの自前ポーリングをやめてSQSイベントソーストリガーに変更した

2018/08/19

以前は、LambdaがSQSからメッセージを受信するには、キューをポーリングしてメッセージを受信する必要がありました。
ポーリングしてキューがあればLambdaを実行するLambdaを動かしたりしていました。

2018年7月にSQSがLambdaのイベントソースとしてサポートされました！
「AWS Lambda がサポートするイベントソースに Amazon Simple Queue Service を追加」

これにより、ポーリングを実装しなくてもよくなりました。

SQSからキューメッセージを受け取ってRedmine APIへデータ登録していたLambdaを改善したいと思います。

本処理をしているLambdaのトリガーをSQSにします。
IAMロールはSQSとCloudWatchに対してのアクセスを許可しました。

目次

event[Records]

eventで渡されるRecordsの中にはすでにキューメッセージも、receiptHandleも含まれています。

{
     "Records": [
        {
            "messageId": "(messageId)",
            "receiptHandle": "(receiptHandle)",
            "body": "",
            "attributes": {
                "ApproximateReceiveCount": "",
                "SentTimestamp": "",
                "SenderId": "",
                "ApproximateFirstReceiveTimestamp": ""
            },
            "messageAttributes": {},
            "md5OfBody": "",
            "eventSource": "aws:sqs",
            "eventSourceARN": "arn:aws:sqs:us-east-1:************:my_queue",
            "awsRegion": "us-east-1"
        }
    ]
}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

{

"Records": [

{

"messageId": "(messageId)",

"receiptHandle": "(receiptHandle)",

"body": "",

"attributes": {

"ApproximateReceiveCount": "",

"SentTimestamp": "",

"SenderId": "",

"ApproximateFirstReceiveTimestamp": ""

},

"messageAttributes": {},

"md5OfBody": "",

"eventSource": "aws:sqs",

"eventSourceARN": "arn:aws:sqs:us-east-1:************:my_queue",

"awsRegion": "us-east-1"

}

]

}

サンプルコード

import logging
import boto3
import json
import traceback
import os

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def lambda_handler(event, context):
    try:
        sqs = boto3.client('sqs')
        sqs_url = os.environ.get('SQS_URL', '')

        for record in event['Records']:
            message_body = json.loads(record['body'])

            #処理

            response = sqs.delete_message(
                QueueUrl=sqs_url,
                ReceiptHandle=record['receiptHandle']
            )
            logger.info(response)

    except:
        raise Exception(traceback.format_exc())

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

import logging

import boto3

import json

import traceback

import os

logger = logging.getLogger()

logger.setLevel(logging.INFO)

def lambda_handler(event, context):

try:

sqs = boto3.client('sqs')

sqs_url = os.environ.get('SQS_URL', '')

for record in event['Records']:

message_body = json.loads(record['body'])

#処理

response = sqs.delete_message(

QueueUrl=sqs_url,

ReceiptHandle=record['receiptHandle']

)

logger.info(response)

except:

raise Exception(traceback.format_exc())

料金

「この機能には追加料金はかかりませんが、Lambda サービスは SQS キューを連続してロングポーリングするため、標準の SQS 価格設定でこれらの API コールに対して課金されます。」とあるので、ポーリングをAWSがやってくれてはいますが、SQSへのリクエスト課金は発生するということですね。

むやみやたらにポーリングしているわけではなく、かしこい頻度調整をしながらロングポーリングをしているようです。
なので自分でコーディングするよりはきっと効率的でしょう。

詳しくは、「AWS Lambda がサポートするイベントソースに Amazon Simple Queue Service を追加」をご参照ください。

これで余分なLambdaも余分なコードも必要なくなりました。

最後までお読みいただきましてありがとうございました！

「AWS認定資格試験テキスト＆問題集　AWS認定ソリューションアーキテクト - プロフェッショナル改訂第2版」という本を書きました。

「AWS認定資格試験テキスト AWS認定クラウドプラクティショナー改訂第3版」という本を書きました。

「ポケットスタディ AWS認定デベロッパーアソシエイト［DVA-C02対応］」という本を書きました。

「要点整理から攻略するAWS認定ソリューションアーキテクト-アソシエイト」という本を書きました。

「AWSではじめるLinux入門ガイド」という本を書きました。

開発ベンダー5年、ユーザ企業システム部門通算9年、ITインストラクター5年目でプロトタイプビルダーもやりだしたSoftware Engineerです。
質問はコメントかSNSなどからお気軽にどうぞ。
出来る限りなるべく答えます。

このブログの内容/発言の一切は個人の見解であり、所属する組織とは関係ありません。
このブログは経験したことなどの共有を目的としており、手順や結果などを保証するものではありません。
ご参考にされる際は、読者様自身のご判断にてご対応をお願いいたします。

また、勉強会やイベントのレポートは自分が気になったことをメモしたり、聞いて思ったことを書いていますので、登壇者の意見や発表内容ではありません。

- AWS AWS, lambda, sqs

Tweet

関連記事

: API Gatewayから直接 DynamoDBに書き込む

やりたいこと WebページでOやXを押したときに、どっちを押したかをDynamo …

: DynamoDBテーブル項目をS3にエクスポート

DynamoDBテーブルを一時テーブルとして使っていて、毎回使い捨てしてます。 …

: Amazon EC2 Auto Scalingのライフサイクルフック

EC2 Auto Scalingにライフサイクルフックという機能があります。ス …

: Amazon Location Service入門ワークショップ-ジオコーディング

ジオコーディングにより住所を緯度経度に変換したり、逆に緯度経度から住所を求めたり …

: AWS Transit GatewayのAcceleratedサイト間VPN接続確認

Transit GatewayでVPN接続を作成するときに、Global Acc …

: AWS CDKでクロススタックリファレンスをする

CloudFormationで複数のスタックで参照することがあります。それをC …

: slackのbotにWikipediaを調べてもらう(Python on AWS Lambda + API Gateway)

slackのbotに少しでも役に立ってもらおうと、Wikipediaを調べてもら …

: Backlogの実績工数をAmazon QuickSightで可視化してわかったこと

今年に入ってから、Backlogで個人タスクを登録しだして、予定工数、実績工数を …

: Amazon Route 53 Resolverを設定確認

Route 53 Resolverを設定しました。東京リージョンのVPCをオン …

: AWS WAFをSIEM on Amazon OpenSearch Serviceで可視化

SIEM on Amazon OpenSearch Serviceを構築の環境に …

PREV: Amazon Data Lifecycle Manager(DLM)が東京リージョンで使えるようになったのでLambdaでAMI自動取得から乗り換えた
NEXT: Redmine起動時にPassengerで「Could not spawn process for application」エラー