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

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

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

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

異步

消息隊列

守護進程

協程池

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

redis

rabbitmq

kafka

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

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

本實例由傳統 MVC 框架投遞郵件發送需求（生產者），Mix PHP 編寫的守護程序執行發送任務（消費者）。

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

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

composer require swiftmailer/swiftmailer

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

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

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

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

通常異步開發中，投遞完成后就會立即響應一個消息給用戶，當然此時該任務并沒有在生產者中執行，而是待消息被消費者獲取后才執行。

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

本例我們采用 Mix PHP V2 的守護程序、協程池來完成一個超高性能的郵件發送程序。

因為我們是開發一個守護程序，所以我們在 applications/daemon 模塊中開發，首先我們在配置 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;

    /**
     * 主函數
     */
    public function main()
    {
        // 捕獲信號
        ProcessHelper::signal([SIGHUP, SIGINT, SIGTERM, SIGQUIT], function ($signal) {
            $this->quit = true;
            ProcessHelper::signal([SIGHUP, SIGINT, SIGTERM, SIGQUIT], null);
        });
        // 協程池執行任務
        xgo(function () {
            $maxWorkers = 20;
            $maxQueue   = 20;
            $jobQueue   = new Channel($maxQueue);
            $dispatch   = new Dispatcher([
                "jobQueue"   => $jobQueue,
                "maxWorkers" => $maxWorkers,
            ]);
            $dispatch->start(MailerWorker::class);
            // 投放任務
            $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); 外部的異常捕獲可得知，當 Redis 連接出錯時，比如 Redis 重啟、連接異常時協程池會安全退出，也就是說當進程異常退出后用戶需使用 supervisor、pm2 等工具重啟守護進程。

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

applications/daemon/src/Libraries/MailerWorker.php

 */
class MailerWorker extends AbstractWorker implements WorkerInterface
{

    /**
     * 郵件發送器
     * @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 類的屬性，當 jobQueue 消息投遞過來時消息會傳遞到 handle 方法，在該方法中使用 Mailer 類的實例完成郵件發送任務，所以我們要編寫了一個 Mailer 發送程序：

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()
    {
        // 開啟協程鉤子
        Coroutine::enableHook();
    }

    /**
     * 發送
     * @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 => "**網"])
            ->setTo($to)
            ->setBody($body);
        // Send the message
        return $mailer->send($message);
    }

}

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

先啟動消費者守護程序：

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

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

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

消費者守護程序結果：

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

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