Yet-another swoole非同期キュー、弾性拡張、ワークプロセス連携サポート

Aint Queue
Swooleベースの非同期キューライブラリでは、伸縮性のあるワークプロセスプールがあり、ワークプロセスのコラボレーションがサポートされています.
Githubアドレス:https://github.com/Littlesqx/...
とくせい

デフォルトRedisドライバ

秒級遅延タスク

カスタム再試行回数および時間

カスタムエラーコールバック

は、タスク実行ミドルウェア

をサポートする.

カスタムキュースナップショットイベント

フレックスマルチプロセス消費

ワークプロセス連携サポート

きれいなダッシュボード

を内蔵
環境

PHP 7.2+

Swoole 4.4+

Redis 3.2+(redis駆動)

インストール

$ composer require littlesqx/aint-queue -vvv

使用
コンフィギュレーション
デフォルトの読み込み構成パス:config/aint-queue.php、存在しない場合/vendor/littlesqx/aint-queue/src/Config/config.phpを読み込みます.

 [...config]
    'default' => [
        'driver' => [
            'class' => RedisQueue::class,
            'connection' => [
                'host' => '127.0.0.1',
                'port' => 6379,
                'database' => '0',
                // 'password' => 'password',
            ],
            'pool_size' => 8,
            'pool_wait_timeout' => 1,
            'handle_timeout' => 60 * 30,
        ],
        'logger' => [
            'class' => DefaultLogger::class,
                'options' => [
                    'level' => \Monolog\Logger::DEBUG,
                ],
            ],
        'pid_path' => '/var/run/aint-queue',
        'consumer' => [
            'sleep_seconds' => 1,
            'memory_limit' => 96,
            'dynamic_mode' => true,
            'capacity' => 6,
            'flex_interval' => 5 * 60,
            'min_worker_number' => 5,
            'max_worker_number' => 30,
            'max_handle_number' => 0,
        ],
        'job_snapshot' => [
            'interval' => 5 * 60,
            'handler' => [],
        ],
    ],
];

すべてのパラメータ:
name
type
comment
default
channel
string
チャンネル.キューの単位は、各チャンネル内のメッセージがそれぞれの消費者と生産者に対応している.マルチチャンネル対応.コマンドラインでは--channelパラメータを使用します.
default
driver.class
string
キュー駆動クラスは、QueueInterfaceを実装する必要があります.
Redis
driver.connection
map
ドライバ構成.
pid_path
string
メインプロセスのPIDファイルストレージパス.注意ユーザーを実行するには、読み取り/書き込み権限が必要です.
/var/run/aint-queue
consumer.sleep_seconds
int
タスクが空いている場合、pop操作後の睡眠秒数.
1
consumer.memory_limit
int
ワークプロセスの最大使用メモリは、超過すると再起動されます.単位MB
96
consumer.dynamic_mode
bool
自動伸縮作業プロセスをオンにするかどうか.
true
consumer.capacity
int
各ワークプロセスが短時間で健康な状態で処理されるメッセージの最大数を表し、ワークプロセスの自動伸縮ポリシーに影響します.
5
consumer.flex_interval
int flex_interval秒ごとに、モニタプロセスは、自動伸縮ワークプロセスがオンになっていると仮定して、ワークプロセスの数を調整しようとします.
5
consumer.min_worker_number
int
ワークプロセスの最小数.
5
consumer.max_worker_number
int
ワークプロセスの最大数.
30
consumer.max_handle_number
int
現在のワークプロセスで処理されているメッセージの最大数が、超過した後に再起動されます.0は無制限です.
0
job_snapshot
map job_snapshot.interval秒ごとに、job_snapshot.handlesが順次実行されます.job_snapshot.handlesはJobSnapshotterInterfaceを実現する必要がある.
メッセージ送信
cli/fpm運転モードで使用できます.

push(function () {
    echo "Hello aint-queue
";
});

// push a delay job
$closureJob = function () {
    echo "Hello aint-queue delayed
";
};
$queue->push($closureJob, 5);

クラスタスクの使用を推奨します.これにより、機能がより完全になり、より良い符号化体験とパフォーマンスが得られます.

作成されたタスククラスはJobInterfaceを実装する必要があります.詳細は/exampleを参照してください.

handle():void:タスクボディ、実行するコンテンツ、ここではswooleの関連api(コパスの作成など)を使用できます.

canRetry(int$attempt,$error):bool:再試行するかどうかを決定する;

retryAfter(int$attempt):int:再試行の遅延時間を決定する;

failed(int$id,array$payload):void:タスクは完全に失敗しました(再試行回数に達しても成功していません).

middleware():array:現在のタスクのミドルウェアリスト、原理はlaravel pipeline流水線を参照します.

タスクは、生産者および消費者において(逆)シーケンス化されなければならないことに注意し、同じプロジェクト

において必要であることを意味する.

キュースナップショットイベントを使用すると、キューのリアルタイム監視を実現できます.タスクミドルウェアを使用すると、タスク実行速度制限、タスク実行ログなどを実現できます.

キュー管理Supervisorなどのプロセス管理ツールを使用して作業プロセスを保護することを推奨します.

vendor/bin/aint-queue

AintQueue Console Tool

Usage:
  command [options] [arguments]

Options:
  -h, --help            Display this help message
  -q, --quiet           Do not output any message
  -V, --version         Display this application version
      --ansi            Force ANSI output
      --no-ansi         Disable ANSI output
  -n, --no-interaction  Do not ask any interactive question
  -v|vv|vvv, --verbose  Increase the verbosity of messages: 1 for normal output, 2 for more verbose output and 3 for debug

Available commands:
  help                 Displays help for a command
  list                 Lists commands
 queue
  queue:clear          Clear the queue.
  queue:dashboard      Start http server for dashboard.
  queue:reload-failed  Reload all the failed jobs onto the waiting queue.
  queue:status         Get the execute status of specific queue.
 worker
  worker:listen        Listen the queue.
  worker:reload        Reload worker for the queue.
  worker:run           Run the specific job.
  worker:stop          Stop listening the queue.

テスト

composer test