使用 mixphp 打造多進程異步郵件發(fā)送

EdwardUp 發(fā)布于2019-07-01 10:12 / 1886人閱讀

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

注意：這個是 MixPHP V1 的范例

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

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

下面演示一個異步郵件發(fā)送系統(tǒng)的開發(fā)過程，涉及知識點：

異步

消息隊列

多進程

守護進程

如何使用消息隊列實現(xiàn)異步

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

redis

rabbitmq

kafka

本次我們選用 redis 來實現(xiàn)異步郵件發(fā)送，redis 的數(shù)據(jù)類型中有一個 list 類型，可實現(xiàn)消息隊列，使用以下命令：

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

架構(gòu)設(shè)計

本實例由傳統(tǒng) MVC 框架投遞郵件發(fā)送需求，MixPHP 多進程執(zhí)行發(fā)送任務。

郵件發(fā)送庫選型

以往我們通常使用框架提供的郵件發(fā)送庫，或者網(wǎng)上下載別的用戶分享的庫，composer 出現(xiàn)后，https://packagist.org/ 上有大量優(yōu)質(zhì)的庫，我們只需選擇一個最好的即可，本例選擇 swiftmailer。

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

composer require swiftmailer/swiftmailer

生產(chǎn)者開發(fā)

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

在傳統(tǒng) 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));

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

消費者開發(fā)

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

TaskExecutor 的 MODE_PUSH 模式有二種進程：

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

中進程：負責執(zhí)行郵件發(fā)送任務。

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();
        // 設(shè)置pidfile
        $this->pidFile = "/var/run/{$this->programName}.pid";
    }

    /**
     * 獲取服務
     * @return TaskExecutor
     */
    public function getTaskService()
    {
        return create_object(
            [
                // 類路徑
                "class"         => "mix	askTaskExecutor",
                // 服務名稱
                "name"          => "mix-daemon: {$this->programName}",
                // 執(zhí)行類型
                "type"          => mix	askTaskExecutor::TYPE_DAEMON,
                // 執(zhí)行模式
                "mode"          => mix	askTaskExecutor::MODE_PUSH,
                // 左進程數(shù)
                "leftProcess"   => 1,
                // 中進程數(shù)
                "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;
    }

    // 左進程啟動事件回調(diào)函數(shù)
    public function onLeftStart(LeftProcess $worker)
    {
        try {
            // 模型內(nèi)使用長連接版本的數(shù)據(jù)庫組件，這樣組件會自動幫你維護連接不斷線
            $queueModel = Redis::getInstance();
            // 保持任務執(zhí)行狀態(tài)，循環(huán)結(jié)束后當前進程會退出，主進程會重啟一個新進程繼續(xù)執(zhí)行任務，這樣做是為了避免長時間執(zhí)行內(nèi)存溢出
            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 出現(xiàn) 100%
            sleep(1);
            // 拋出錯誤
            throw $e;
        }
    }

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

    // 發(fā)送郵件
    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 => "**網(wǎng)"])
            ->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.

調(diào)用接口往消息隊列投放任務。

此時 shell 終端將打印：

成功收到測試郵件：

MixPHP

GitHub: https://github.com/mixstart/m...
官網(wǎng)：http://www.mixphp.cn/

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

文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/28918.html

發(fā)表評論

登陸后可評論

0條評論

EdwardUp

男|高級講師

我要關(guān)注我要私信

TA的文章

云主機怎么賣-云虛擬主機如何購買？買云虛擬主機需要注意哪些事項？

閱讀 2918·2021-09-22 15:20
云主機怎么上傳網(wǎng)站-怎么在云主機上搭建網(wǎng)站？

閱讀 2991·2021-09-22 15:19
主機過期域名怎么辦-網(wǎng)站域名到期后被搶注怎么辦？

閱讀 3506·2021-09-22 15:15
涂鴉智能 ① tuya-wifi-mcu-sdk-arduino-library 簡單剖析

閱讀 2433·2021-09-08 09:35
css學習day05

閱讀 2408·2019-08-30 15:44
VueCli+Node+mongodb打造個人博客（含前臺展示及后臺管理系統(tǒng)）（下）

閱讀 3042·2019-08-30 10:50
前端每日實戰(zhàn)：109# 視頻演示如何用 CSS 和 D3 創(chuàng)作一個用文字組成的心形圖案

閱讀 3792·2019-08-29 16:25
CSS3實現(xiàn)炫酷的切片式圖片輪播效果

閱讀 1616·2019-08-26 13:55

成人国产在线小视频_日韩寡妇人妻调教在线播放_色成人www永久在线观看_2018国产精品久久_亚洲欧美高清在线30p_亚洲少妇综合一区_黄色在线播放国产_亚洲另类技巧小说校园_国产主播xx日韩_a级毛片在线免费

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

使用 mixphp 打造多進程異步郵件發(fā)送

相關(guān)文章

**Mix PHP V2 實例：協(xié)程池異步郵件發(fā)送守護程序**

**我為什么要開發(fā)一個 MixPHP 框架**

**使用 mixphp 處理其他框架 20% 的高并發(fā)部分**

MixPHP：基于 Swoole 的常駐內(nèi)存型 PHP 框架

MixPHP 環(huán)境搭建之 Apache + PHP

發(fā)表評論

0條評論

EdwardUp

男|高級講師

TA的文章

云主機怎么賣-云虛擬主機如何購買？買云虛擬主機需要注意哪些事項？

云主機怎么上傳網(wǎng)站-怎么在云主機上搭建網(wǎng)站？

主機過期域名怎么辦-網(wǎng)站域名到期后被搶注怎么辦？

涂鴉智能 ① tuya-wifi-mcu-sdk-arduino-library 簡單剖析

css學習day05

VueCli+Node+mongodb打造個人博客（含前臺展示及后臺管理系統(tǒng)）（下）

前端每日實戰(zhàn)：109# 視頻演示如何用 CSS 和 D3 創(chuàng)作一個用文字組成的心形圖案

CSS3實現(xiàn)炫酷的切片式圖片輪播效果

最新活動

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

使用 mixphp 打造多進程異步郵件發(fā)送

相關(guān)文章

發(fā)表評論

0條評論

男|高級講師

TA的文章

最新活動

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！