Amazon Simple Queue Service(SQS)をPHPでいじる #AWS

SQSはシンプルなキューサービスです。料金も良心的で(ほぼ0)、実装上の面倒な部分を上手いことスケーリング出来ます。

実際DBとかで実装して実装出来ない事もないんですが、スケーリング時の排他制御とかめちゃくちゃ面倒なので、こういうのは外部サービスにお任せするのが一番かと。

[Getting Start] http://docs.aws.amazon.com/aws-sdk-php/guide/latest/service-sqs.html
[詳細API] http://docs.aws.amazon.com/aws-sdk-php/latest/class-Aws.Sqs.SqsClient.html

PHPでの導入

Composerに対応してます。

composer require aws/aws-sdk-php:dev-master

クライアントの作成

$auth = [
    "key" => "AWS_AUTH_KEY",
    "secret" => "AWS_AUTH_SECRET",
    "region" => "us-west-2",
];
$sqs = SqsClient::factory($auth);

region抜きで認証入れちゃうと以下みたいな感じで怒られちゃうんですが、利用可能なリージョンのリストも教えてくれます。ツンデレです。可愛い。

Uncaught exception Aws\Common\Exception\InvalidArgumentException: A region is required when using Amazon Simple Queue Service. Set "region" to one of: us-east-1, us-west-1, us-west-2, eu-west-1, ap-northeast-1, ap-southeast-1, ap-southeast-2, sa-east-1, cn-north-1, us-gov-west-1
Callstack:

キューの作成

クライアントが作成できたら次はキューを作成します。
キューはメッセージを入れるための箱みたいなもので、メッセージの読み書きをする場合には、このキューのURLが必要になります。
あ、別にAWSコンソール上からやってもいいです。

$result = $$sqs->createQueue(array(
    'QueueName' => 'string',
    'Attributes' => array(
        QueueAttribute::DELAY_SECONDS        => 5,
));
$queueUrl = $result->get('QueueUrl');

QueueNameは必須で、Attributesはオプション。オプションの一覧/詳細とかは、AWSコンソール上でキューの作成試してみるとわかりやすいです。

キューを作成したら、getメソドでQueueUrlが取得可能。
ちなみに同一名称のキューを作成しても何も起きません。

メッセージの送信

メッセージはキューに対して送信します。

$sqs->sendMessage(array(
    'QueueUrl'    => $queueUrl,
    'MessageBody' => 'An awesome !'.date("H:i:s"),
));

メッセージの受信

キューに送信したメッセージを読み取るには,receiveMessageを使用します。

$result = $sqs->receiveMessage(array(
    'QueueUrl' => $queueUrl,
));

キューは一度読み取ると、他の接続から連続して読み取る事が出来ません(排他制御)。デフォルトでは30秒の排他制御時間が設定されており、キューの設定、QueueAttribute::VISIBILITY_TIMEOUTで変更する事も出来ます。

読み取り時にオプションを使用する事も可能です。WaitTimeSecondsはメッセージが見つからない場合にロングポーリングします。

$result = $client->receiveMessage(array(
    'QueueUrl'        => $queueUrl,
    'WaitTimeSeconds' => 10,
));

メッセージはデフォルトで一件づつ取り出されますが、MaxNumberOfMessagesで複数のメッセージをまとめて取り出すことも可能です。

$result = $client->receiveMessage(array(
    'QueueUrl'        => $queueUrl,
    'MaxNumberOfMessages' => 10,
));

メッセージのパース

ドキュメント見るとパースはgetPathメソドを使ってやる感じになってるんですが、これがちょっとつかいづらい感じです。

$result = $sqs->receiveMessage(array(
    'QueueUrl' => $queueUrl,
));

foreach ($result->getPath('Messages/*/Body') as $messageBody) {
    echo $messageBody;
}

ドキュメントにはこんな感じで書いてますが、これ、メッセージが無い時getPathの戻り値がnullになって死にます。
あとBodyしか取れない。

getPathのパスを一個戻してやると、だいたいこんな感じの値が取れます。

$result = $sqs->receiveMessage(array(
            'QueueUrl' => $queueUrl,
        ));
$res = $result->getPath('Messages/*');
var_dump($res);

array(4) {
  'Body' =>
  string(20) "An awesome !15:57:22"
  'MD5OfBody' =>
  string(32) "59d6c3279f516e8bcaf972347da65f8e"
  'ReceiptHandle' =>
  string(300) "uUk89DY........"
  'MessageId' =>
  string(36) "7718309d-512f-4185-a9ee-90bebb901c4c"
}

Body: メッセージの内容
MD5OfBody: ↑のMD5
ReceiptHandle: メッセージのハンドルキー。削除する時とかに必要
MessageId: 何これ

やっぱりメッセージない時はNULLが返ってきます。

でも、複数件受信とかするとBodyとかMD5OfBodyとかそれぞれのキーの下に配列ができるのでやっぱり使いづらい。どうしたものか。

削除

キューの削除は、deleteQueueメソドを使います。

$sqs->deleteQueue(["QueueUrl"=>$queueUrl]);

一度削除したキューと同名のキューは60秒間の間作成することが出来ません。エラーになります。注意してください。

メッセージを削除する場合には、ReceiptHandleが必要です。
受診する度に値が変わるので受信したタイミングで削除する必要があります。

$sqs->deleteMessage([
    "QueueUrl" => $queueUrl,
    "ReceiptHandle" => $res["ReceiptHandle"]
]);