AMAZON SQS(一)PHP Producer

AMAZON SQS(1)PHP プロデューサー
AMAZON SQS(1)PHP プロデューサー

1.基本情報
価格: 100 万 2 ドル
送信スループット:
受信スループット:
メッセージ遅延: 500 ミリ秒
メッセージ サイズ: 256 KB
バッチ サイズ: 10 メッセージ

2.環境設定
コンポーザーのインストール
>カール -sS https://getcomposer.org/installer | php

最新のパッケージをインストール
> php combos.phar require aws/aws-sdk-php

これにより、すべての依存関係がインストールされます
>。 php combos.phar install

依存関係を識別するためのcomposer.jsonは次のとおりです。
{
"require": {
"aws/aws-sdk-php": "3.0 .6",
"predis/predis": "1.0.1"
}
}

そして、以下のコマンドは、Maven または Ivy のような魔法を実行します
>php combos.phar install

一部のインポート クラスは次のようになります。
SQSConnector.inc

require 'vendor/autoload.php';

use AwsSqsSqsClient;

class SQSConnector {
const queueURL = "https://sqs.us-east-1.amazonaws.com/xx_test";

protected $client = null;

function __construct(){
try {
$this->client = SqsClient::factory(array(
'region) ' => 'us-east-1',
'version' => 'latest'
));
echo "SQSn に正常に接続されました";
} catch (Exception $e ) {
echo "SQS に接続できませんでした";
echo $e->getMessage();
}
}
}

タイム トラック クラスTimeTracker.inc

class TimeTrackerimplements Countable {

protected $count = 0;
これにより、プロセス中にかかる時間の詳細が得られます。 protected $distribution;
protected $distributionPrecision = 0;
protected $startTime;
protected $totalTime;
protected $name;

public function __construct($name, $distributionPrecision = 2) {
$this->name = $name;
$this->distribution = array();
$this->distributionPrecision = $distributionPrecision;
$this-> ;start();
}

public function count() {
return $this->count;
}

public function start($time = null ) {
return $this->startTime = ($time === null) ? microtime(true) : $time;
}

public function stop($count = 1, $time = null) {
$interval = (($time === null) ? microtime (true) : $time) - $this->startTime;
$this->totalTime = $interval;
$key = (string)round($interval, $this->distributionPrecision);
if (!array_key_exists($key, $this->distribution)) $this->distribution[$key] = 0;
$this->distribution[$key] ;
$ this->count = $count;
return $interval;
}

public function __toString() {
ksort($this->distribution);
// "{$this->name}: " を返します。 print_r($this->distribution, true) .
return "{$this->name}: {$this->count}m " 。 round($this->totalTime,2) 。の ' .
round($this->count/$this->totalTime,1) 。 "m/sn";
}

}

非同期 SQSSender.php
#!/usr/bin/php

require_once 'TimeTracker.inc';
require_once 'SQSConnector.inc';
use GuzzleHttpPromise;

define('MSGNUM_REQUEST', 10);
define( 'MAXSIZE_REQUEST', 262144);

class SQSSender extends SQSConnector {
function sendMessage($job) {
$timer = new TimeTracker('sendMessage');
$promise_array = array( );

if(strlen($job) >= MAXSIZE_REQUEST)
{
echo "メッセージのサイズが 256KBn を超えています";
var_dump($job);
return;
}

$promise = $this->client->sendMessageAsync(array(
'QueueUrl' => self::queueURL,
'MessageBody' = > $job,
));
$promise->wait();
$timer->stop();

print $timer;
}

function sendMessageBatch($jobs, $timer) {
$promise_array = array();


$id=1;
$message = '';
$msg_num = 0;
foreach($jobs as $job)
{
$maxsize_msg = (int)(MAXSIZE_REQUEST/MSGNUM_REQUEST);
$tmp_msg = $message.$job;
if(strlen($job) >= $maxsize_msg)
{
$this->sendMessage($job);
}
else if(strlen($tmp_msg) >= $maxsize_msg)
{
$entries[] = array(
'Id' => $id,
'メッセージ本文' => $message
);
$id ;
$message = $job;
if($id > 10){
$promise = $this->client->sendMessageBatchAsync(array(
'QueueUrl' => self::queueURL) ,
'エントリ' => $entries,
));
$entries = array();
$id = 1;

$promise_array['key' . count($timer)] = $promise;
/*if(count($promise_array) % 50 == 0){
Promiseunwrap($promise_array); //Promise リストのラップを解除し、すべての Promise が完了するまで待ちます
$promise_array = array();
}*/

$timer->stop($msg_num);
$msg_num = 0;
$timer->start();
}
$msg_num ;
}
else
{
$message = $tmp_msg;
$ msg_num ;
}
}
$entries[] = array(
'Id' => $id,
'MessageBody' => $message
);
$promise = $this->client->sendMessageBatchAsync(array(
'QueueUrl' => self::queueURL,
'Entries' => $entries,
));
$promise_array['key' . count($timer)] = $promise;
Promiseunwrap($promise_array); //Promise リストのラップを解除し、すべての Promise が完了するまで待ちます
$promise_array = array();
$timer->stop($msg_num);
print $timer;
$timer-> ;start();
}
}

/*$sender = new SQSSender;
$promise = $sender->send(500);
if(count( $promise) > 0){
Promiseunwrap($promises);
}*/

?>

テストを実行する前に、AMAZON は次の設定を要求します。ここのキー>
carl-mac:.aws carl$ pwd
/Users/carl/.aws
carl-mac:.aws carl$ ls -l
total 8
-rw -r--r-- 1 carl Staff 116 Jul 6 17:00 credentials
carl-mac:.aws carl$

内容は次のようになります:
[デフォルト]
aws_access_key_id = xxxxxxxx
aws_secret_access_key = xxxxxxxxxxxxx

参考文献:
http://colby.id.au/benchmarking-sqs/
http://www.warski.org/blog/2014 /06/benchmarking-sqs/
http://nsono.net/amazon-sqs-vs-rabbitmq/

kafka
http://engineering.linkedin.com/kafka/benchmarking- apache-kafka-2-million-writes-second-three-cheap-machines

SQS
http://docs.aws.amazon.com/aws-sdk-php/guide/latest/service -sqs.html
http://aws.amazon.com/documentation/sqs/
http://docs.aws.amazon.com/aws-sdk-php/v3/guide/guide/promises。 html

http://docs.aws.amazon.com/aws-sdk-php/v3/api/index.html

scala キュー
https://github.com /adamw/mqperf/tree/master/src/main/scala/com/softwaremill/mqperf/mq

SQS 自動スケーリング
https://aws.amazon.com/articles/Amazon-SQS/ 1464