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

去年 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

<?php

namespace Daemon\Commands;

use Daemon\Libraries\MailerWorker;
use Mix\Concurrent\CoroutinePool\Dispatcher;
use Mix\Core\Coroutine\Channel;
use Mix\Helper\ProcessHelper;

/**
 * Class MailerCommand
 * @package Daemon\Commands
 * @author liu,jian <coder.keda@gmail.com>
 */
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

<?php

namespace Daemon\Libraries;

use Mix\Concurrent\CoroutinePool\AbstractWorker;
use Mix\Concurrent\CoroutinePool\WorkerInterface;

/**
 * Class MailerWorker
 * @package Daemon\Libraries
 * @author liu,jian <coder.keda@gmail.com>
 */
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

<?php

namespace Daemon\Libraries;

use Mix\Core\Coroutine;

/**
 * Class Mailer
 * @package Daemon\Libraries
 * @author liu,jian <coder.keda@gmail.com>
 */
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

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