摘要：消費(fèi)者開發(fā)使用本例時，請確保你使用的編譯時開啟了本例我們采用的守護(hù)程序協(xié)程池來完成一個超高性能的郵件發(fā)送程序。

去年 Mix PHP V1 發(fā)布時，我寫了一個多進(jìn)程的郵件發(fā)送實例: 使用 mixphp 打造多進(jìn)程異步郵件發(fā)送，今年 Mix PHP V2 發(fā)布，全面的協(xié)程支持讓我們可以使用一個進(jìn)程就可達(dá)到之前多個進(jìn)程都無法達(dá)到的更高 IO 性能，所以今天重寫一個協(xié)程池版本的郵件發(fā)送實例。

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

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

異步

消息隊列

守護(hù)進(jìn)程

協(xié)程池

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);

本實例由傳統(tǒng) MVC 框架投遞郵件發(fā)送需求（生產(chǎn)者），Mix PHP 編寫的守護(hù)程序執(zhí)行發(fā)送任務(wù)（消費(fèi)者）。

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

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

composer require swiftmailer/swiftmailer

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

在傳統(tǒng) MVC 框架的控制器中增加如下代碼：

通?？蚣苤惺褂?Redis 會安裝一個類庫來使用，本例使用原生代碼，便于理解。

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

通常異步開發(fā)中，投遞完成后就會立即響應(yīng)一個消息給用戶，當(dāng)然此時該任務(wù)并沒有在生產(chǎn)者中執(zhí)行，而是待消息被消費(fèi)者獲取后才執(zhí)行。

使用本例時，請確保你使用的 Swoole 編譯時開啟了 openssl

本例我們采用 Mix PHP V2 的守護(hù)程序、協(xié)程池來完成一個超高性能的郵件發(fā)送程序。

因為我們是開發(fā)一個守護(hù)程序，所以我們在 applications/daemon 模塊中開發(fā)，首先我們在配置 applications/daemon/config/main.php 中注冊一個命令：

// 命令
"commands"         => [

    "mailer" => ["Mailer", "description" => "Mailer daemon."],

],

注冊的命令中指定的 Mailer 命令類，接下來我們編寫一個 MailerCommand 類：

applications/daemon/src/Commands/MailerCommand.php

 */
class MailerCommand
{

    /**
     * 退出
     * @var bool
     */
    public $quit = false;

    /**
     * 主函數(shù)
     */
    public function main()
    {
        // 捕獲信號
        ProcessHelper::signal([SIGHUP, SIGINT, SIGTERM, SIGQUIT], function ($signal) {
            $this->quit = true;
            ProcessHelper::signal([SIGHUP, SIGINT, SIGTERM, SIGQUIT], null);
        });
        // 協(xié)程池執(zhí)行任務(wù)
        xgo(function () {
            $maxWorkers = 20;
            $maxQueue   = 20;
            $jobQueue   = new Channel($maxQueue);
            $dispatch   = new Dispatcher([
                "jobQueue"   => $jobQueue,
                "maxWorkers" => $maxWorkers,
            ]);
            $dispatch->start(MailerWorker::class);
            // 投放任務(wù)
            $redis = app()->redisPool->getConnection();
            while (true) {
                if ($this->quit) {
                    $dispatch->stop();
                    return;
                }
                try {
                    $data = $redis->brPop(["queue:email"], 3);
                } catch (Throwable $e) {
                    $dispatch->stop();
                    return;
                }
                if (!$data) {
                    continue;
                }
                $data = array_pop($data); // brPop命令最后一個鍵才是值
                $jobQueue->push($data);
            }
        });
    }

}

從 $data = $redis->brPop(["queue:email"], 3); 外部的異常捕獲可得知，當(dāng) Redis 連接出錯時，比如 Redis 重啟、連接異常時協(xié)程池會安全退出，也就是說當(dāng)進(jìn)程異常退出后用戶需使用 supervisor、pm2 等工具重啟守護(hù)進(jìn)程。

上面是一個 Mix PHP 協(xié)程池的使用代碼，基本可以直接復(fù)制使用，框架默認(rèn)包含了協(xié)程池的 Demo，本次實例只是修改了協(xié)程池的 Worker，本命令主要是完成從 Redis 隊列中獲取消息然后 push 到 jobQueue 中，jobQueue 中的數(shù)據(jù)會被 20 個 Worker 實例中某一個搶占后并行執(zhí)行，本例的郵件發(fā)送代碼邏輯就在 MailerWorker 類中：

applications/daemon/src/Libraries/MailerWorker.php

 */
class MailerWorker extends AbstractWorker implements WorkerInterface
{

    /**
     * 郵件發(fā)送器
     * @var Mailer
     */
    public $mailer;

    /**
     * 初始化事件
     */
    public function onInitialize()
    {
        parent::onInitialize(); // TODO: Change the autogenerated stub
        // 實例化一些需重用的對象
        $this->mailer = new Mailer();
    }

    /**
     * 處理
     * @param $data
     */
    public function handle($data)
    {
        // TODO: Implement handle() method.
        $data = unserialize($data);
        if (empty($data)) {
            return;
        }
        try {
            $this->mailer->send($data["to"], $data["subject"], $data["body"]);
            app()->log->info("Mail sent successfully：to {to} subject {subject}", $data);
        } catch (Throwable $e) {
            app()->log->error("Mail failed to send：to {to} subject {subject} error {error}", array_merge($data, ["error" => $e->getMessage()]));
        }
    }

}

由以上代碼可見，Worker 在初始化時，新增了一個 Mailer 類的屬性，當(dāng) jobQueue 消息投遞過來時消息會傳遞到 handle 方法，在該方法中使用 Mailer 類的實例完成郵件發(fā)送任務(wù)，所以我們要編寫了一個 Mailer 發(fā)送程序：

applications/daemon/src/Libraries/Mailer.php

 */
class Mailer
{

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

    /**
     * Mailer constructor.
     */
    public function __construct()
    {
        // 開啟協(xié)程鉤子
        Coroutine::enableHook();
    }

    /**
     * 發(fā)送
     * @param $to
     * @param $subject
     * @param $body
     * @return int
     */
    public function send($to, $subject, $body)
    {
        // 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($subject))
            ->setFrom([self::USERNAME => "**網(wǎng)"])
            ->setTo($to)
            ->setBody($body);
        // Send the message
        return $mailer->send($message);
    }

}

在 Mailer 發(fā)送程序中我們使用了前面 composer 安裝的 swiftmailer 庫來發(fā)送郵件，以上就完成了全部的代碼邏輯，現(xiàn)在我們開始測試。

先啟動消費(fèi)者守護(hù)程序：

[root@localhost bin]# ./mix-daemon mailer

將上文的生產(chǎn)者腳本命名為 push.php 然后在 CLI 中執(zhí)行 (開一個新終端)：

[root@localhost bin]# php /tmp/push.php

消費(fèi)者守護(hù)程序結(jié)果：

[root@localhost bin]# ./mix-daemon mailer
[info] 2019-04-15 11:48:36 [message] Mail sent successfully：to ***@qq.com subject The title content

命令行終端打印了發(fā)送成功的日志，發(fā)送完成。