使用 mixphp 打造多進程異步郵件發送

EdwardUp 發布于2019-07-01 10:12 / 1858人閱讀

摘要：消費者開發本例我們使用的多進程開發工具來完成這個需求，通常使用常駐進程來處理隊列的消費，所以我們使用的類型，模式。中進程負責執行郵件發送任務。此時終端將打印成功收到測試郵件官網

注意：這個是 MixPHP V1 的范例

郵件發送是很常見的需求，由于發送郵件的操作一般是比較耗時的，所以我們一般采用異步處理來提升用戶體驗，而異步通常我們使用消息隊列來實現。

傳統 MVC 框架由于缺少多進程開發能力，通常是采用同一個腳本執行多次，產生多個進程的方式，mixphp 封裝了 TaskExecutor 專用于多進程開發，用戶能非常簡單的開發出功能完善的高可用多進程應用。

下面演示一個異步郵件發送系統的開發過程，涉及知識點：

異步

消息隊列

多進程

守護進程

如何使用消息隊列實現異步

PHP 使用消息隊列通常是使用中間件來實現，常用的消息中間件有：

redis

rabbitmq

kafka

本次我們選用 redis 來實現異步郵件發送，redis 的數據類型中有一個 list 類型，可實現消息隊列，使用以下命令：

// 入列
$redis->lpush($key, $data);
// 出列
$data = $redis->rpop($key);
// 阻塞出列
$data = $redis->brpop($key, 10);

架構設計

本實例由傳統 MVC 框架投遞郵件發送需求，MixPHP 多進程執行發送任務。

郵件發送庫選型

以往我們通常使用框架提供的郵件發送庫，或者網上下載別的用戶分享的庫，composer 出現后，https://packagist.org/ 上有大量優質的庫，我們只需選擇一個最好的即可，本例選擇 swiftmailer。

由于發送任務是由 MixPHP 執行，所以 swiftmailer 是安裝在 MixPHP 項目中，在項目根目錄中執行以下命令安裝：

composer require swiftmailer/swiftmailer

生產者開發

在郵件發送這個需求中生產者是指投遞發送任務的一方，這一方通常是一個接口或網頁，這個部分并不一定需 mixphp 開發，TP、CI、YII 這些都可以，只需在接口或網頁中把任務信息投遞到消息隊列中即可。

在傳統 MVC 框架的控制器中增加如下代碼：

通常框架中使用 redis 會安裝一個類庫來使用，本例使用原生代碼，便于理解。

// 連接
$redis = new Redis();
if (!$redis->connect("127.0.0.1", 6379)) {
    throw new Exception("Redis Connect Failure");
}
$redis->auth("");
$redis->select(0);
// 投遞任務
$data = [
    "to"      => ["***@qq.com" => "A name"],
    "body"    => "Here is the message itself",
    "subject" => "The title content",
];
$redis->lpush("queue:email", serialize($data));

通常異步開發中，投遞完成后就會立即響應一個消息給用戶，當然此時該任務并沒有執行。

消費者開發

本例我們使用 MixPHP 的多進程開發工具 TaskExecutor 來完成這個需求，通常使用常駐進程來處理隊列的消費，所以我們使用 TaskExecutor 的 TYPE_DAEMON 類型，MODE_PUSH 模式。

TaskExecutor 的 MODE_PUSH 模式有二種進程：

左進程：負責從消息隊列取出任務數據，投放給中進程。

中進程：負責執行郵件發送任務。

PushCommand.php 代碼如下：


 */
class PushCommand extends BaseCommand
{

    // 配置信息
    const HOST = "smtpdm.aliyun.com";
    const PORT = 465;
    const SECURITY = "ssl";
    const USERNAME = "****@email.***.com";
    const PASSWORD = "****";

    // 初始化事件
    public function onInitialize()
    {
        parent::onInitialize(); // TODO: Change the autogenerated stub
        // 獲取程序名稱
        $this->programName = Input::getCommandName();
        // 設置pidfile
        $this->pidFile = "/var/run/{$this->programName}.pid";
    }

    /**
     * 獲取服務
     * @return TaskExecutor
     */
    public function getTaskService()
    {
        return create_object(
            [
                // 類路徑
                "class"         => "mix	askTaskExecutor",
                // 服務名稱
                "name"          => "mix-daemon: {$this->programName}",
                // 執行類型
                "type"          => mix	askTaskExecutor::TYPE_DAEMON,
                // 執行模式
                "mode"          => mix	askTaskExecutor::MODE_PUSH,
                // 左進程數
                "leftProcess"   => 1,
                // 中進程數
                "centerProcess" => 5,
                // 任務超時時間 (秒)
                "timeout"       => 5,
            ]
        );
    }

    // 啟動
    public function actionStart()
    {
        // 預處理
        if (!parent::actionStart()) {
            return ExitCode::UNSPECIFIED_ERROR;
        }
        // 啟動服務
        $service = $this->getTaskService();
        $service->on("LeftStart", [$this, "onLeftStart"]);
        $service->on("CenterStart", [$this, "onCenterStart"]);
        $service->start();
        // 返回退出碼
        return ExitCode::OK;
    }

    // 左進程啟動事件回調函數
    public function onLeftStart(LeftProcess $worker)
    {
        try {
            // 模型內使用長連接版本的數據庫組件，這樣組件會自動幫你維護連接不斷線
            $queueModel = Redis::getInstance();
            // 保持任務執行狀態，循環結束后當前進程會退出，主進程會重啟一個新進程繼續執行任務，這樣做是為了避免長時間執行內存溢出
            for ($j = 0; $j < 16000; $j++) {
                // 從消息隊列中間件阻塞獲取一條消息
                $data = $queueModel->brpop("queue:email", 10);
                if (empty($data)) {
                    continue;
                }
                list(, $data) = $data;
                // 將消息推送給中進程去處理，push有長度限制 (https://wiki.swoole.com/wiki/page/290.html)
                $worker->push($data, false);
            }
        } catch (Exception $e) {
            // 休息一會，避免 CPU 出現 100%
            sleep(1);
            // 拋出錯誤
            throw $e;
        }
    }

    // 中進程啟動事件回調函數
    public function onCenterStart(CenterProcess $worker)
    {
        // 保持任務執行狀態，循環結束后當前進程會退出，主進程會重啟一個新進程繼續執行任務，這樣做是為了避免長時間執行內存溢出
        for ($j = 0; $j < 16000; $j++) {
            // 從進程消息隊列中搶占一條消息
            $data = $worker->pop();
            if (empty($data)) {
                continue;
            }
            // 處理消息
            try {
                // 處理消息，比如：發送短信、發送郵件、微信推送
                var_dump($data);
                $ret = self::sendEmail($data);
                var_dump($ret);
            } catch (Exception $e) {
                // 回退數據到消息隊列
                $worker->rollback($data);
                // 休息一會，避免 CPU 出現 100%
                sleep(1);
                // 拋出錯誤
                throw $e;
            }
        }
    }

    // 發送郵件
    public static function sendEmail($data)
    {
        // Create the Transport
        $transport = (new Swift_SmtpTransport(self::HOST, self::PORT, self::SECURITY))
            ->setUsername(self::USERNAME)
            ->setPassword(self::PASSWORD);
        // Create the Mailer using your created Transport
        $mailer = new Swift_Mailer($transport);
        // Create a message
        $message = (new Swift_Message($data["subject"]))
            ->setFrom([self::USERNAME => "**網"])
            ->setTo($data["to"])
            ->setBody($data["body"]);
        // Send the message
        $result = $mailer->send($message);
        return $result;
    }

}

測試

在 shell 中啟動 push 常駐程序。

[root@localhost bin]# ./mix-daemon push start
mix-daemon "push" start successed.

調用接口往消息隊列投放任務。

此時 shell 終端將打印：

成功收到測試郵件：

MixPHP

GitHub: https://github.com/mixstart/m...
官網：http://www.mixphp.cn/

云服務器 GPU云服務器 Django下使用celery 異步發送阿里云服務器Linux使用PHP郵件發送 php 異步進程使用深度學習打造

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/28918.html

Mix PHP V2 實例：協程池異步郵件發送守護程序

摘要：消費者開發使用本例時，請確保你使用的編譯時開啟了本例我們采用的守護程序協程池來完成一個超高性能的郵件發送程序。去年 Mix PHP V1 發布時，我寫了一個多進程的郵件發送實例: 使用 mixphp 打造多進程異步郵件發送，今年 Mix PHP V2 發布，全面的協程支持讓我們可以使用一個進程就可達到之前多個進程都無法達到的更高 IO 性能，所以今天重寫一個協程池版本的郵件發送實例。...

lauren_liuling 2019-07-01 12:41 評論0 收藏0
我為什么要開發一個 MixPHP 框架

摘要：異步隊列消費者開發只提供了模式下運行控制器方法，并未提供主進程多子進程的進程模型，并未提供多線程處理。多線程異步隊列服務只需寫好控制器方法，然后在配置文件中配置下路由命名空間進程線程數量，就可在模式中啟動多進程多線程模型的異步隊列處理程序。最近業余時間一直在開發ExpressPHP的第二個版本 MixPHP，今天下班想起之前一個面試官的問題：你為什么還要再造一個輪子呢？仔細回想，第一...

SnaiLiu 2019-06-28 14:14 評論0 收藏0
使用 mixphp 處理其他框架 20% 的高并發部分

摘要：如何使用優化高并發場景寫庫或者耗時計算在的接口中使用消息隊列，把要入庫的數據寫入的類型中。高容錯子進程異常奔潰時，主進程將重建子進程。高性能多進程運行，充分利用多個并行計算，性能強勁。經常在群里聽到一些朋友問：TP 的項目怎么遷移到 mixphp 來處理高并發，我通常都是回復需要重寫，可是一個開發很久的 TP 項目，代碼量巨大，又怎么可能會花大量時間成本來重寫呢？那么為何我們不嘗試...

silvertheo 2019-07-01 12:26 評論0 收藏0
MixPHP：基于 Swoole 的常駐內存型 PHP 框架

摘要：框架最新源代碼行數行，因此可以很容易的改造它，成為你們公司的專屬框架。也不同于其他基于的微服務框架，只聚焦于微服務治理，定位于開發的更多領域，覆蓋從初創到億元級體量的技術訴求。的授權全靠用戶自愿購買，詳情 MixPHP是什么 MixPHP 是秉承普及 PHP 常駐內存型解決方案，促進 PHP 往更后端發展的理念而創造，采用 Swoole 擴展作為底層引擎，圍繞常駐內存的方式而設計，...

王晗 2019-07-01 12:25 評論0 收藏0
MixPHP 環境搭建之 Apache + PHP

摘要：在多種環境中遷移，代碼無需修改，是無縫遷移的。由于大部分用戶開發是在中進行，因此開發階段我們推薦使用部署方案，因為更簡單快速，下面整體演示一下的環境搭建。安裝解壓至指定安裝目錄。先不要啟動，這會啟動會報錯，沒加環境變量。 MixPHP 是一款基于 Swoole 的常駐內存型 PHP 高性能框架。 MixPHP 同時支持多種環境中執行： Nginx + mix-httpd （使用到 S...

Barrior 2019-07-24 10:44 評論0 收藏0