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

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

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

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

異步

消息隊(duì)列

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

協(xié)程池

PHP 使用消息隊(duì)列通常是使用中間件來(lái)實(shí)現(xiàn)，常用的消息中間件有：

redis

rabbitmq

kafka

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

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

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

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

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

composer require swiftmailer/swiftmailer

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

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

通?？蚣苤惺褂?Redis 會(huì)安裝一個(gè)類庫(kù)來(lái)使用，本例使用原生代碼，便于理解。

// 連接
$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));

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

使用本例時(shí)，請(qǐng)確保你使用的 Swoole 編譯時(shí)開(kāi)啟了 openssl

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

因?yàn)槲覀兪情_(kāi)發(fā)一個(gè)守護(hù)程序，所以我們?cè)?applications/daemon 模塊中開(kāi)發(fā)，首先我們?cè)谂渲?applications/daemon/config/main.php 中注冊(cè)一個(gè)命令：

// 命令
"commands"         => [

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

],

注冊(cè)的命令中指定的 Mailer 命令類，接下來(lái)我們編寫一個(gè) MailerCommand 類：

applications/daemon/src/Commands/MailerCommand.php

 */
class MailerCommand
{

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

    /**
     * 主函數(shù)
     */
    public function main()
    {
        // 捕獲信號(hào)
        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命令最后一個(gè)鍵才是值
                $jobQueue->push($data);
            }
        });
    }

}

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

上面是一個(gè) Mix PHP 協(xié)程池的使用代碼，基本可以直接復(fù)制使用，框架默認(rèn)包含了協(xié)程池的 Demo，本次實(shí)例只是修改了協(xié)程池的 Worker，本命令主要是完成從 Redis 隊(duì)列中獲取消息然后 push 到 jobQueue 中，jobQueue 中的數(shù)據(jù)會(huì)被 20 個(gè) Worker 實(shí)例中某一個(gè)搶占后并行執(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
        // 實(shí)例化一些需重用的對(duì)象
        $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()]));
        }
    }

}

由以上代碼可見(jiàn)，Worker 在初始化時(shí)，新增了一個(gè) Mailer 類的屬性，當(dāng) jobQueue 消息投遞過(guò)來(lái)時(shí)消息會(huì)傳遞到 handle 方法，在該方法中使用 Mailer 類的實(shí)例完成郵件發(fā)送任務(wù)，所以我們要編寫了一個(gè) 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()
    {
        // 開(kāi)啟協(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 庫(kù)來(lái)發(fā)送郵件，以上就完成了全部的代碼邏輯，現(xiàn)在我們開(kāi)始測(cè)試。

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

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

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

[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ā)送完成。