async-helper，一個 PHP 的異步進程助手

時間 2019-12-11

標籤 async helper 一個 php 異步進程助手欄目 PHP 简体版

原文原文鏈接

async-helper

簡介

PHP 的異步進程助手，藉助於 AMQP 實現異步執行 PHP 的方法，將一些很耗時、追求高可用、須要重試機制的操做放到異步進程中去執行，將你的 HTTP 服務從繁重的業務邏輯中解脫出來。以一個較低的成本將傳統 PHP 業務邏輯轉換成非阻塞、高可用、可擴展的異步模式。php

依賴

php 5.6+
ext-bcmath
ext-amqp 1.9.1+
ext-memcached 3.0.3+

安裝

經過 composer 安裝git

composer require l669/async-helper

或直接下載項目源碼github

wget https://github.com/l669306630/async-helper/archive/master.zip

使用範例

業務邏輯：這裏定義了不少等待被調用的類和方法，在你的項目中這多是數據模型、或是一個發送郵件的類。數據庫

<?php
class SendMailHelper 
{
    /**
     * @param array $mail
     * @throws Exception
     */
    public static function request($mail)
    {
        // 在這裏發送郵件，或是經過調用第三方提供的服務發送郵件
        // 發送失敗的時候你拋出了異常，但願被進程捕獲，並按設定的規則進行重試
    }    
}

生產者：一般是 HTTP 服務，傳統的 PHP 項目或是一個命令行程序，接收到某個請求或指令後進行一系列的操做。segmentfault

<?php 
use l669\AsyncHelper;
class UserController
{
    public function register()
    {
        // 假設這是一個用戶註冊的請求，用戶提交了姓名、郵箱、驗證碼
        // 第一步、校驗用戶信息
        // 第二步、實例化異步助手，這時候會鏈接 AMQP
        $async_helper = new AsyncHelper([
            'host' => '127.0.0.1',
            'port' => '5672',
            'user' => 'root',
            'pass' => '123456',
            'vhost' => '/'
        ]);
        // 第三步、保存用戶信息到數據庫
        $mail = [
            'from' => 'service@yourdomain.com', 
            'to' => 'username@163.com', 
            'subject' => '恭喜你註冊成功',
            'body' => '請點擊郵件中的連接完成驗證....'
        ];
        // 第四步、經過異步助手發送郵件
        $async_helper->run('\\SendMailHelper', 'request', [$mail]);
        
        // 這是同步的模式去發送郵件，若是郵件服務響應遲緩或異常，就會直接影響該請求的響應時間，甚至丟失這封重要郵件
        // SendMailHelper::request($mail);
    }
}

消費者：PHP 的異步進程，監聽消息隊列，執行你指定的方法。而且該消費者進程是可擴展的高可用的服務，這一切都得益於 AMQP，這是系統解耦、佈局微服務的最佳方案。後端

consume.php瀏覽器

<?php
require_once('vendor/autoload.php');
require_once('SendMailHelper.php');

use l669\AsyncHelper;
use l669\CacheHelper;

$cache_helper = new CacheHelper('127.0.0.1', 11211);
while(true){
    try{
        $async_helper = new AsyncHelper([
            'host' => '127.0.0.1',
            'port' => '5672',
            'user' => 'root',
            'pass' => '123456',
            'vhost' => '/',
            'cacheHelper' => $cache_helper
        ]);
        $async_helper->consume();
    }catch(Exception $e){
        // 能夠在這裏記錄一些日誌
        sleep(2);
    }
}

# 在命令行下啓動消費者進程，推薦使用 supervisor 來管理進程
php consume.php

支持事務：須要一次提交執行多個異步方法，事務能夠確保完成性。服務器

// 接着上面的示例來講，這裏省略了一些重複的代碼，下同
$async_helper->beginTransaction();
try{
    $async_helper->run('\\SendMailHelper', 'request', [$mail1]);
    $async_helper->run('\\SendMailHelper', 'request', [$mail2]);
    $async_helper->run('\\SendMailHelper', 'request', [$mail3]);
    $async_helper->commit();
}catch(\Exception $e){
    $async_helper->rollback();
}

阻塞式重試：當異步進程執行一個方法，方法內部拋出異常時進行重試，一些必須遵循執行順序的業務就要採用阻塞式的重試，經過指定重試最大阻塞時長來控制。微信

use l669\CacheHelper;
use l669\AsyncHelper;
$async_helper = new AsyncHelper([
    'host' => '127.0.0.1',
    'port' => '5672',
    'user' => 'root',
    'pass' => '123456',
    'vhost' => '/',
    'cacheHelper' => new CacheHelper('127.0.0.1', 11211),
    'retryMode' => AsyncHelper::RETRY_MODE_REJECT,  // 阻塞式重試
    'maxDuration' => 600                            // 最長重試 10 分鐘
]);
$send_mail_helper = new \SendMailHelper();
$mail = new \stdClass();
$mail->from = 'service@yourdomain.com';
$mail->to = 'username@163.com';
$mail->subject = '恭喜你註冊成功';
$mail->body = '請點擊郵件中的連接完成驗證....';
$async_helper->run($send_mail_helper, 'request', [$mail]);

// 若是方法中須要拋出異常來結束程序，又不但願被異步進程重試，能夠拋出如下幾種錯誤碼，進程捕獲到這些異常後會放棄重試：
// l669\AsyncException::PARAMS_ERROR
// l669\AsyncException::METHOD_DOES_NOT_EXIST
// l669\AsyncException::KNOWN_ERROR

非阻塞式重試：當異步執行的方法內部拋出異常，async-helper 會將該方法從新放進隊列的尾部，先執行新進入隊列的方法，回頭再重試剛纔執行失敗的方法，經過指定最大重試次數來控制。php7

use l669\CacheHelper;
use l669\AsyncHelper;
$async_helper = new AsyncHelper([
    'host' => '127.0.0.1',
    'port' => '5672',
    'user' => 'root',
    'pass' => '123456',
    'vhost' => 'new',
    'cacheHelper' => new CacheHelper('127.0.0.1', 11211),
    'queueName' => 'emails.vip',                    // 給付費的大爺走 VIP 隊列
    'retryMode' => AsyncHelper::RETRY_MODE_TTL,     // 非阻塞式重試
    'maxRetries' => 10                              // 最多重試 10 次
]);
$mail = new \stdClass();
$mail->from = 'service@yourdomain.com';
$mail->to = 'username@163.com';
$mail->subject = '恭喜你註冊成功';
$mail->body = '請點擊郵件中的連接完成驗證....';
$async_helper->run('\\SendMailHelper', 'request', [$mail]);

應用和解惑

咱們採用的是開源的 RabbitMQ 來爲咱們提供的 AMQP 服務。
你的項目部署在擁有不少服務器節點的集羣上，每一個節點的程序都須要寫日誌文件，如今的問題就是要收集全部節點上面的日誌到一個地方，方便咱們及時發現問題或是作一些統計。全部節點均可以使用 async-helper 異步調用一個寫日誌的方法，而執行這個寫日誌的方法的進程只須要在一臺機器上啓動就能夠了，這樣全部節點的日誌就都實時掌握在手裏了。
作過微信公衆號開發的都知道，騰訊微信能夠將用戶的消息推送到咱們的服務器，若是咱們在 5s 內未及時響應，騰訊微信會重試 3 次，其實這就是消息隊列的應用，使用 async-helper 能夠輕鬆的作和這同樣的事情。
得益於 RabbitMQ，你能夠輕鬆的橫向擴展你的消費者進程的能力，由於 RabbitMQ 天生就支持集羣部署，你能夠輕鬆的啓動多個消費者進程，或是將消費者進程分佈到多臺機器上。
若是 RabbitMQ 服務不可用怎麼辦呢？部署 RabbitMQ 高可用服務是容易的，對外提供單一 IP，這個 IP 是個負載均衡，背後是 RabbitMQ 集羣，負載均衡承擔對後端集羣節點的健康檢查。
async-helper 可否承受高併發請求？async-helper 生產者使用的是短鏈接，也就說在你的 HTTP 尚未響應瀏覽器的時候 async-helper 就已經結束了工做，你鏈接 RabbitMQ 的時間是百分之百小於 HTTP 請求的時間的，換言之，只要 RabbitMQ 承受併發的能力超過你的 HTTP 服務的承受併發的能力，RabbitMQ 就永遠不會崩，經過橫向擴展 RabbitMQ 很容易作到的。