FlowerMQ

FlowerMQ 一個(gè)基于Workerman和Redis實(shí)現(xiàn)的消息隊(duì)列,一個(gè)小小工具，用來(lái)給主項(xiàng)目解耦的，也支持延遲隊(duì)列，失敗嘗試這些。

運(yùn)行依賴

• php7.2
• Redis5.0.4上，因?yàn)橛玫絉edis Stream
• pecl依賴，redis擴(kuò)展
• composer依賴，workerman/workerman 4.0以上

安裝

composer安裝

composer create-project mrtwenty/flower

下載安裝

1. 下載或者 git clone
2. 項(xiàng)目根目錄執(zhí)行命令，composer install

原理說(shuō)明

1. workerman實(shí)現(xiàn)消費(fèi)端，開(kāi)多個(gè)進(jìn)程，在 onWorkerStart 函數(shù)里面，阻塞讀取，阻塞間隔5秒后，就重新阻塞,因?yàn)槭亲枞?，所以用了一個(gè)字符串key來(lái)處理停止的問(wèn)題，每次阻塞5秒，就斷開(kāi)，以便判斷是否需要終止程序。
2. 可以隨時(shí)停止消費(fèi)端，因?yàn)榭蛻舳税l(fā)送的消息都會(huì)存放redis stream 隊(duì)列里面。
3. 一個(gè)pending進(jìn)程,每隔1秒檢查是否有未ack的消息，并嘗試消費(fèi)掉
4. 一個(gè)delay進(jìn)程,負(fù)責(zé)處理延遲消息，利用redis的zset有序集合存儲(chǔ)，起一個(gè)定時(shí)器，定時(shí)獲取可以執(zhí)行的消息，寫(xiě)入消費(fèi)端
5. 遵循約定大于配置的方式，直接用默認(rèn)的即可。
6. 默認(rèn)配置是app目錄下的config目錄，如果需要更改配置項(xiàng)，可以在項(xiàng)目根目錄下，提供一個(gè).env的配置文件，替換掉
7. 回收裁剪機(jī)制: 有三種模式，默認(rèn) no ，不做裁剪
   1. no,不做裁剪，所有消息保留。
   2. maxlen, 最大長(zhǎng)度回收,概率性觸發(fā) xtrim maxlen mq ~ 長(zhǎng)度 。
   3. minid, 最小已讀消息回收，概率行觸發(fā)，xtrim minid mq ~ 消息id ，需要 redis server 6.2.0 以上。

可用命令

win

windows下僅限于開(kāi)發(fā)，不適合做生產(chǎn)環(huán)境使用，啟動(dòng)需要開(kāi)三個(gè)命令行窗口，執(zhí)行 start、pending、delay命令

1. php index.php start 啟動(dòng)消費(fèi)隊(duì)列
2. php index.php pending 啟動(dòng)重試隊(duì)列
3. php index.php delay 啟動(dòng)延遲隊(duì)列
4. php index.php test 測(cè)試，執(zhí)行此命令會(huì)發(fā)送兩個(gè)消息給服務(wù)，一個(gè)是即時(shí)消息，一個(gè)是延遲消息。
5. php monitor.php start 運(yùn)行信息查看，會(huì)啟動(dòng)一個(gè)http進(jìn)程

linux

1. php index.php start linux啟動(dòng)相當(dāng)于執(zhí)行了 start、pending、delay命令
2. php index.php start -d 守護(hù)進(jìn)程啟動(dòng)
3. php index.php stop 強(qiáng)制停止,可能會(huì)導(dǎo)致消息未ack，不建議使用
4. php index.php stop -g 優(yōu)雅停止 (不加參數(shù)-g會(huì)強(qiáng)制干掉子進(jìn)程，加參數(shù)-g的話，會(huì)等子進(jìn)程處理完后再關(guān)閉)
5. php index.php config 查看配置信息
6. php index.php test 測(cè)試，執(zhí)行此命令會(huì)發(fā)送兩個(gè)消息給服務(wù)，一個(gè)是即時(shí)消息，一個(gè)是延遲消息。
7. php monitor.php start 運(yùn)行信息查看，會(huì)啟動(dòng)一個(gè)http進(jìn)程

服務(wù)端說(shuō)明

1. 下載項(xiàng)目后，配置 .env
2. 編寫(xiě)業(yè)務(wù)邏輯，app\consumer\Run.php 只需要編寫(xiě)這里，如果代碼有curl請(qǐng)求，記得要做好超時(shí)
3. 啟動(dòng)，php index.php start 即可。

客戶端說(shuō)明

flower配備了一個(gè)客戶端，方便在別的項(xiàng)目中使用:

composer install mrtwenty/flower-client

使用方式:

$redis = new Redis;
$redis->connect('127.0.0.1', 6379);
//$mq需要與服務(wù)端三個(gè)配置信息相同
$mq    = [
    'name' => 'mq',
    'delay_name' => 'mq_delay',
    'fail_list' => 'mq_fail_list'
];
$client = new Client($redis, $mq);

//立即執(zhí)行
$res = $client->add(['test' => 'data']);
var_dump($res);
//延遲消息
$res = $client->add(['test' => 'data'], 3);
var_dump($res);

問(wèn)題

1.引入MySQL

可以安裝此依賴包，當(dāng)然也可以根據(jù)自己需要用別的包

composer require workerman/mysql

配置信息可以在.env里面寫(xiě)入:

[mysql]
host = 127.0.0.1
username = root
password = 123456
database = test
port     = 3306

代碼實(shí)現(xiàn):

<?php

declare(strict_types=1);

namespace app;

use app\library\BaseInterface;

/**
 * 消費(fèi)類
 */
class Run implements BaseInterface
{
    protected $db = null;

    public function getDb()
    {
        if (is_null($this->db)) {
            $config = config('mysql');
            $host     = $config['host'];
            $port     = $config['port'];
            $user     = $config['username'];
            $password = $config['password'];
            $database  = $config['database'];
            $this->db = new \Workerman\MySQL\Connection($host, $port, $user, $password, $database);
        }
        return $this->db;
    }

    /**
     * 消費(fèi)方法，如何消費(fèi)，取決用戶自己
     *
     * @param mixed $data
     * @param mixed $id
     * @return bool 返回true就會(huì)執(zhí)行ack確認(rèn)消息已消費(fèi)
     */
    public function consumer($data, $id): bool
    {
        $db   = $this->getDb();
        $info = $db->row("SELECT * FROM `short_url` WHERE id=3");
        print_r($info);
        return true;
    }

    /**
     * 超過(guò)嘗試的次數(shù)，就會(huì)寫(xiě)入失敗隊(duì)列里面，并調(diào)用此方法，可以用此方法通知運(yùn)維
     *
     * @return void
     */
    public function fail($data, $id)
    {
        print_r($data);
        print_r($id);
    }
}

2. 避免內(nèi)存泄露

由于是守護(hù)進(jìn)程，為了避免php業(yè)務(wù)代碼bug隱藏的內(nèi)存泄露，可以在消費(fèi)者執(zhí)行完一定數(shù)量的時(shí)候重啟進(jìn)程。具體實(shí)現(xiàn)請(qǐng)查看workerman手冊(cè)。

鏈接1、鏈接2

相關(guān)資料

1. redis stream 手冊(cè)

2. pecl redis 文檔

3. workerman 手冊(cè)

引用

1. monitor登錄頁(yè)模板
2. env、config類這些學(xué)自thinkphp
3. monitor 后端的一些代碼，學(xué)自webman
4. monitor主頁(yè)面，用的layui
5. 延遲隊(duì)列的思路抄的workerman的redis queue
6. 感謝workerman、thinkphp、layui、redis