centos（php7）下gearman實現異步處理隊列任務

安裝gearman須要的環境

wget https://github.com/libevent/libevent/releases/download/release-2.1.8-stable/libevent-2.1.8-stable.tar.gz
tar zxvf libevent-2.1.8-stable.tar.gz
cd libevent-2.1.8-stable
./configure --prefix=/usr
make
make install

yum install libevent-devel -y
yum install boost boost-devel -y
yum install gperf* -y
yum install libuuid-devel -y

安裝gearman

wget https://github.com/gearman/gearmand/releases/download/1.1.17/gearmand-1.1.17.tar.gz
tar zxvf gearmand-1.1.17.tar.gz
cd gearmand-1.1.17
./configure --prefix=/usr/local/gearman
make
make install
/sbin/ldconfig

gearman自動

mikdir /usr/local/gearman/var/log
gearmand -d -l /usr/local/gearman/var/log/gearman.log

gearmand -V
# 輸出：gearmand 1.1.17 - https://github.com/gearman/gearmand/issues

安裝php擴展gearman

wget http://pecl.php.net/get/gearman-1.1.2.tgz
#若是是php7：wget https://github.com/wcgallego/pecl-gearman/archive/master.zip
tar zxvf gearman-1.1.2.tgz
cd gearman-1.1.2
/usr/local/php/bin/phpize
./configure --with-php-config=/usr/local/php/bin/php-config
make
make install

修改php.ini

extension = gearman.so
#重啓php-fpm 或者 重啓apache

程序實現

簡單的的字符反轉案例（來自官方翻譯，稍做修改）

client.php 任務發送方

<?php

$client = new GearmanClient();

// 添加服務器
// php官方文檔：http://php.net/manual/en/gearmanclient.addserver.php
// $client->addServer("10.0.0.2", 7003);
$client->addServer();

echo "發送任務: \n";

// 發送 {"msg":"test"}, workload消息給 task_func_name 任務
$data = '{"msg":"' . $argv[1] . '"}';
$result = $client->doNormal("task_func_name", $data);      //普通任務，阻塞
#$result = $client->doBackground("task_func_name", $data); //後臺任務，直接返回任務句柄
var_dump($result);

?>

work.php 任務接收消費方

<?php

$worker = new GearmanWorker();

$worker->addServer() or die("沒法連接服務器");

// 通知服務器，$worker綁定task_func_name處理函數work_func。
$worker->addFunction("task_func_name", "work_func");

while (1) {
    print "接收任務...\n";
    $ret = $worker->work(); // reverse_fn 任務執行完成
    if ($ret != true) { //代碼和下面同樣
        break;
    }
    //獲取最後一次執行任務的返回碼 執行成功正常返回0
    if ($worker->returnCode() != GEARMAN_SUCCESS) {
        break;
    }
}

// A much simple reverse function
function work_func(GearmanJob $job) {
    $workload = $job->workload(); //取得發送方發送的消息
    echo "Received job: " . $job->handle() . "\n";  //獲取任務的句柄，有服務器自動分配，對用戶不透明
    echo "Workload: $workload\n";
    $result = json_decode($workload, true);
    $result['msg'] = strrev($result['msg']);
    $json = json_encode($result);
    echo "Result: $json\n";
    return $json;
}

?>

調用方式：

#啓動接收消費方
php work.php
#output: 接收任務...

//啓動發送消息方
php client.php testmsg1
# output: 
# 發送任務:
# string(18) "{"msg":"1gsmtset"}"
php client.php testmsg2
# output: 
# 發送任務:
# string(18) "{"msg":"2gsmtset"}"

修改client裏改爲後臺
php client.php testmsg1
# output: 
# string(26) "H:localhost.localdomain:59" //這裏再也不返回的結果了

總結

無論是阻塞仍是異步直接返回，消費者這邊都是根據入隊列的順序依次執行，即：一個執行完再接一個，若是一個任務消費很長時間，後續任務都會在等待前個處理完才能夠繼續執行

多任務案列（來自官方，小刀改）

client.php

<?php
$client = new GearmanClient();
$client->addServer();

// 初始化三個變量
$userInfo = $friends = $posts = null;

//消費端執行完成任務這裏會被異步調用
$client->setCompleteCallback(function(GearmanTask $task, $context) use (&$userInfo, &$friends, &$posts) {
    switch($context) {
        case 'content_task1':
            $userInfo = $task->data();
            $aa = $task->data();
            break;
        case 'content_task2':
            $friends = $task->data();
            $aa = $task->data();
            break;
        case 'content_task3':
            $posts = $task->data();
            $aa = $task->data();
            break;
    }
});

// 添加任務
$client->addTask('task1', 'qkl', 'content_task1');
$client->addTask('task2', 'qkl', 'content_task2');
$client->addTask('task3', 'qkl', 'content_task3');

echo "讀取中...\n";
$start = microtime(true);
$client->runTasks();
$totaltime = number_format(microtime(true) - $start, 2);

echo "獲得結果消耗的時間: $totaltime 秒:\n";
var_dump($userInfo, $friends, $posts);
?>

work.php

<?php

$worker = new GearmanWorker();
$worker->addServer();

//這裏注意，多任務的執行順序是根據消費者這邊註冊消息類型的順序依次來執行，且要等上一次執行完成才能夠繼續
//如今註冊順序是task1 task2 task3 執行順序就是task1 task2 task3
//如今註冊順序是task3 task2 task1 執行順序就是task3 task2 task1
$worker->addFunction('task1', function(GearmanJob $job){
    sleep(2);
    return 'task1: '. $job->workload();
});

$worker->addFunction('task2', function(GearmanJob $job){
    return 'task2: '. $job->workload();
});

$worker->addFunction('task3', function(GearmanJob $job){
    return 'task3: '. $job->workload();
});

while (true) {
    $ret = $worker->work();
    var_dump($ret);
};

?>

總結

這裏多任務場景目前沒有特別好的場景構想，理論那些一步步同步執行順序的流程的業務，只是gearman能夠client添加任務後直接執行後續任務，業務交給消費端慢慢執行