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

関連記事

: AWS CodeStarで静的webサイトのテンプレートプロジェクトを作成する

執筆環境の検討中です。 CodeCommitは使うつもりで、コミットしたときにE …

: AWS Lambda(Python)からAmazon Connectで電話を発信する

自動で電話を発信する必要がありまして、電話発信APIを開発することになりましたの …

: Amazon Quantum Ledger Database(QLDB)でサンプル台帳の作成と検証

Quantum Ledger Database(QLDB)を触ったことなかったの …

: CloudFrontのカスタムヘッダーがなければALBのルーティングで403レスポンスを返す

大阪リージョンにはWAFがまだないです(2021年４月現在) 今のこのブログの構 …

: AWS Cloud9でJavaサンプルを実行する

リモートで共有開発ができるCloud9便利ですね。 Cloud9でJavaのサン …

: AWS Managed Microsoft ADを構築してユーザー追加まで

事前準備 DNSホスト名と名前解決を有効にしたVPCを作成して、2つのAZにパブ …

: kintoneでEveryoneに権限が設定されているアプリをAWS Lambdaで一括チェックする

こないだ、kintone Cafeでユーザーが自由に作成している環境だと、どんな …

: SCPが影響しないサービスにリンクされたロールにEC2が引き受けるIAMロールは含まれないことを確認

ドキュメントで確認サービスコントロールポリシーのユーザーガイドには、「SCPは …

: EC2ユーザーデータからメタデータを取得してRocket.Chatで80ポートを使用する

Rocket.Chatのデフォルトポート番号は3000です。 80を使うようにす …

: boto3(Python SDK) s3 get_object でバイト範囲を指定する

S3のGetObjectアクションでバイト範囲を指定することができます。それに …

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