<h1 align="center">Playcat Queue</h1>

<p align="center">php消息隊列服務</p>

特點

1. 支持多種消息隊列(Redis,Kafka)
2. 延遲消息機制
3. 易于使用和擴展

支持的消息隊列

Redis單機(已完成)

Redis集群(已完成)

Kafka(todo)

更多。。。

環(huán)境需求

• PHP >= 7.2
• Redis >= 5.0

安裝

本項目基于webman,所以請先安裝好webman在執(zhí)行下面操作。

$ composer require "playcat/queue"

使用

1.選擇合適的消息服務端

• 使用redis單機(默認)
  修改config\plugin\playcat\queue\redis.php為自己redis的配置即可

• 使用redis集群機
  修改config\plugin\playcat\queue\manager.php里的driver為如下內容

  'driver' => \Playcat\Queue\Driver\Rediscluster::class,

  修改config\plugin\playcat\queue\rediscluster.php為自己redis的配置即可

2.創(chuàng)建消費者任務

編輯'config/plugin/playcat/queue/redis.php',修改對應的redis配置

新建一個名為'Test.php'文件添加以下內容:

<?php

namespace app\queue\playcat;

use Playcat\Queue\Model\Payload;
use Playcat\Queue\Protocols\Consumer;

class Test implements Consumer
{
    //任務名稱
    public $queue = 'test';

    public function consume(Payload $payload)
    {
        //獲取自定義傳入的內容
        $data = $payload->getQueueData();
        ...
    }
}

將'Test.php'保存到'

app/queue/playcat/'目錄下。如果目錄不存在就創(chuàng)建它(==可以編輯config/plugin/playcat/queue/process.php中的consumer_dir的地址來改變==)

啟動webman的服務

$ php start.php start

添加任務

use Playcat\Queue\Manager;
use Playcat\Queue\Model\Payload;
$payload = new Payload();
//對應消費隊列里的任務名稱
$payload->setChannel('test');
//對應消費隊列里的任務使用的數據
$payload->setQueueData([1,2,3,4]);
//創(chuàng)建一個立即執(zhí)行的任務
Manager::getInstance()->push($payload);

$payload_delay = new Payload();
//對應消費隊列里的任務名稱
$payload_delay->setChannel('test');
//對應消費隊列里的任務使用的數據
$payload_delay->setQueueData([6,7,8,9]);
//設置60秒后執(zhí)行的任務
$payload_delay->setDelayTime(60);
Manager::getInstance()->push($payload_delay);`

異常與重試機制

任務在執(zhí)行過程中未拋出異常則默認執(zhí)行成功，否則則進入重試階段.
重試次數和時間由配置控制，重試間隔時間為當前重試次數的冪函數。
Playcat\Queue\Exceptions\DontRetry異常會忽略掉重試

Playcat\Queue\Model\Payload

• getID: 當前任務的唯一id
• getRetryCount(): 當前任務已經重試過的次數
• getQueueData(): 當前任務傳入的參數
• getChannel(): 當前所執(zhí)行的任務名稱

License

MIT