AmazonSQSの使用

インストール

Amazon SQSサポートの場合、追加の依存関係をインストールする必要があります。 celery[sqs] bundle を使用して、Celeryとこれらの依存関係の両方を一度にインストールできます。

$ pip install celery[sqs]

構成

ブローカーURLでSQSを指定する必要があります。

broker_url = 'sqs://ABCDEFGHIJKLMNOPQRST:ZYXK7NiynGlTogH8Nj+P9nlE73sq3@'

ここで、URL形式は次のとおりです。

sqs://aws_access_key_id:aws_secret_access_key@

常に正しく解析できるように、最後に@記号を含め、パスワードをエンコードすることを忘れないでください。例えば：

from kombu.utils.url import safequote

aws_access_key = safequote("ABCDEFGHIJKLMNOPQRST")
aws_secret_key = safequote("ZYXK7NiynG/TogH8Nj+P9nlE73sq3")

broker_url = "sqs://{aws_access_key}:{aws_secret_key}@".format(
    aws_access_key=aws_access_key, aws_secret_key=aws_secret_key,
)

ログインクレデンシャルは、環境変数 AWS_ACCESS_KEY_IDおよび AWS_SECRET_ACCESS_KEYを使用して設定することもできます。その場合、ブローカーのURLは [のみです。 X157X]。

インスタンスでIAMロールを使用している場合は、BROKER_URLをsqs://に設定すると、kombuはインスタンスメタデータからアクセストークンを取得しようとします。

オプション

領域

デフォルトのリージョンはus-east-1ですが、：setting： `broker_transport_options` 設定を構成することで、別のリージョンを選択できます。

broker_transport_options = {'region': 'eu-west-1'}

も参照してください

アマゾンウェブサービスリージョンの概要はここにあります：

http://aws.amazon.com/about-aws/globalinfrastructure/

可視性タイムアウト

可視性タイムアウトは、メッセージが別のワーカーに再配信される前に、ワーカーがタスクを確認するのを待機する秒数を定義します。以下の警告も参照してください。

このオプションは、：setting： `broker_transport_options` 設定を介して設定されます：

broker_transport_options = {'visibility_timeout': 3600}  # 1 hour.

デフォルトの可視性タイムアウトは30分です。

ポーリング間隔

ポーリング間隔は、失敗したポーリング間でスリープする秒数を決定します。この値は、intまたはfloatのいずれかです。デフォルトでは、値は 1秒です。これは、読み取るメッセージがなくなると、ワーカーが1秒間スリープすることを意味します。

ポーリングの頻度が高いほどコストも高くなるため、ポーリング間隔を長くするとコストを節約できることに注意する必要があります。

ポーリング間隔は、：setting： `broker_transport_options` 設定で設定できます。

broker_transport_options = {'polling_interval': 0.3}

ポーリング間隔が非常に頻繁になると、ビジーループが発生し、ワーカーが多くのCPU時間を使用する可能性があります。サブミリ秒の精度が必要な場合は、次のような別のトランスポートの使用を検討する必要があります。 RabbitMQ 、また Redis 。

長いポーリング

SQSロングポーリングはデフォルトで有効になっており、 ReceiveMessage 操作のWaitTimeSecondsパラメーターは10秒に設定されています。

WaitTimeSecondsパラメーターの値は、：setting： `broker_transport_options` 設定を介して設定できます。

broker_transport_options = {'wait_time_seconds': 15}

有効な値は0〜20です。新しく作成されたキュー自体（Celeryによって作成された場合も）のデフォルト値は、「メッセージ待機時間の受信」キュープロパティに設定されていることに注意してください。

キュープレフィックス

デフォルトでは、Celeryはキュー名にプレフィックスを割り当てません。SQSを使用する他のサービスがある場合は、：setting： `broker_transport_options` 設定を使用して設定できます。

broker_transport_options = {'queue_name_prefix': 'celery-'}

事前定義されたキュー

CeleryがAWSで事前定義されたキューのセットを使用し、SQSキューを一覧表示したり、作成または削除したりしないようにする場合は、：setting： `predefined_queues`を使用してキュー名のマップをURLに渡します。設定：

broker_transport_options = {
    'predefined_queues': {
        'my-q': {
            'url': 'https://ap-southeast-2.queue.amazonaws.com/123456/my-q',
            'access_key_id': 'xxx',
            'secret_access_key': 'xxx',
        }
    }
}

バックオフポリシー

バックオフポリシーは、タスクの再試行間の時間差を変更するSQS可視性タイムアウトメカニズムを使用しています。このメカニズムは、メッセージ固有のvisibility timeoutをキューDefault visibility timeoutからポリシー設定のタイムアウトに変更します。再試行の回数はSQS（具体的にはApproximateReceiveCountメッセージ属性）によって管理され、ユーザーはそれ以上のアクションを行う必要はありません。

キューとバックオフポリシーの構成：

broker_transport_options = {
    'predefined_queues': {
        'my-q': {
            'url': 'https://ap-southeast-2.queue.amazonaws.com/123456/my-q',
            'access_key_id': 'xxx',
            'secret_access_key': 'xxx',
            'backoff_policy': {1: 10, 2: 20, 3: 40, 4: 80, 5: 320, 6: 640},
            'backoff_tasks': ['svc.tasks.tasks.task1']
        }
    }
}

backoff_policyキーは再試行回数、値は再試行間の遅延秒数（つまり、SQS可視性タイムアウト）の辞書backoff_tasks上記のポリシーを適用するタスク名のリスト

上記のポリシー：

試み	遅れ
`2nd attempt`	20秒
`3rd attempt`	40秒
`4th attempt`	80秒
`5th attempt`	320秒
`6th attempt`	640秒

STSトークン認証

https://docs.aws.amazon.com/cli/latest/reference/sts/assume-role.html

AWS STS認証は、sts_role_arnおよびsts_token_timeoutブローカートランスポートオプションを使用してサポートされます。 sts_role_arnは、SQSへのアクセスを承認するために使用する想定されるIAMロールARNです。 sts_token_timeoutはトークンのタイムアウトであり、デフォルト（および最小）は900秒です。上記の期間が経過すると、新しいトークンが作成されます。

ブローカー_トランスポート_オプション= {

'predefined_queues'：{

'my-q'：{

'url'： ' https://ap-southeast-2.queue.amazonaws.com/123456/my-q '、 'access_key_id'： 'xxx'、 'secret_access_key'： 'xxx'、 'backoff_policy'：{1：10、2：20、3：40、4：80、5：320、6：640}、 'backoff_tasks'：['svc.tasks.tasks.task1']

}

},

'sts_role_arn'： 'arn：aws：iam :: ：role / STSTest '、＃オプション' sts_token_timeout '：900＃オプション}

警告

visibility_timeout内でタスクが確認応答されない場合、タスクは別のワーカーに再配信されて実行されます。
これにより、実行時間が可視性タイムアウトを超えるETA /カウントダウン/再試行タスクで問題が発生します。実際、それが発生した場合、それは再び実行され、ループで再び実行されます。

したがって、使用する予定の最長のETAの時間に一致するように、可視性のタイムアウトを増やす必要があります。

Celeryはワーカーのシャットダウン時にメッセージを再配信するため、可視性のタイムアウトが長いと、停電や強制終了したワーカーが発生した場合にのみ、「失われた」タスクの再配信が遅れることに注意してください。

定期的なタスクは、ETA /カウントダウンとは別の概念であるため、可視性タイムアウトの影響を受けません。

この記事の執筆時点でAWSがサポートしている最大可視性タイムアウトは12時間（43200秒）です。
broker_transport_options = {'visibility_timeout': 43200}
SQSは、ワーカーのリモートコントロールコマンドをまだサポートしていません。
SQSはまだイベントをサポートしていないため、セロリイベント、セロリモン、またはDjangoAdminモニターでは使用できません。

結果

アマゾンウェブサービスファミリーの複数の製品は、結果を保存または公開するのに適した候補になる可能性がありますが、現時点ではそのような結果バックエンドは含まれていません。

警告

amqp結果バックエンドをSQSで使用しないでください。

タスクごとに1つのキューが作成され、キューは収集されません。これには、AWS結果ストアのバックエンドをCeleryに戻すために費やしたほうがよい費用がかかる可能性があります:)

Amazon SQSの使用—Pythonドキュメント

目次