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認定資格試験テキスト　AWS認定AIプラクティショナー」という本を書きました。

「AWS教科書 AWS認定デベロッパーアソシエイトテキスト＆問題集」という本を書きました。

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

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

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

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

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

- AWS AWS, lambda, sqs

Tweet

関連記事

: Amazon LinuxにAlfresco Community Editionをインストールする

ダウンロード準備 Alfresco公式サイトで、「エンタープライズコンテンツ管理 …

: RDS自動バックアップの自動クロスリージョンコピー

このブログのRDS for MySQLはAWS Backupで大阪リージョンから …

: RDS + VPC + Lambda + API Gateway + CloudFront + WAF + ACMでAPIを構築する

RDSのMySQLの情報を与えられたリクエストをキーにしてjsonで返すAPIを …

: プライベートサブネットのEC2でセッションマネージャを使うようVPCエンドポイントを構成する

インターネットゲートウェイへのルートがないルートテーブルに関連付けられたプライベ …

: JAWS DAYS 2018 「Cost-Driven AWS クラウドアーキテクチャデザインとコスト最適化方法 – Cost-Driven AWS Cloud Architecture Design : The Lean Startup on AWS」を聞きました

以下は、思ったことや気になったことをメモしていますので、必ずしも登壇者の発表内容 …

: 試そうとしてたらSavings Plans買っちゃいました

Savings Plansの購入画面を確認していました。画面遷移も確認しようと …

: CloudWatchエンドポイントがIPv6に対応したのでCloudWatchエージェントからカスタムメトリクスを送信しました

追記翌日のブログで、CloudWatchエージェントのアップデートにより、Cl …

: WordPress W3 Total Cache のDatabaseCacheをAmazon ElastiCacheのmemcachedに格納する

このブログのアーキテクチャは現在こちらです。データベースは、Amazon Au …

: Projectタグを強制するSCPでアクションやリソースを限定しないとき成り立つのか

ふと疑問に思ったので試しました。予想は、拒否されてしまう、です。 [crayo …

: kintone webhookからAWS API Gateway – Lambdaを実行しレコードの値を渡す

2017年2月のアップデートでkintoneにWebhook機能がリリースされま …

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