教你使用mixphp打造多进程异步邮件发送

composer require swiftmailer/swiftmailer

生产者开发

在邮件发送这个需求中生产者是指投递发送任务的一方，这一方通常是一个接口或网页，这个部分并不一定需 mixphp 开发，TP、CI、YII 这些都可以，只需在接口或网页中把任务信息投递到消息队列中即可。

在传统 MVC 框架的控制器中增加如下代码：

通常框架中使用 redis 会安装一个类库来使用，本例使用原生代码，便于理解。

// 连接
$redis = new \Redis();
if (!$redis->connect(&＃39;127.0.0.1&＃39;, 6379)) {
    throw new \Exception(&＃39;Redis Connect Failure&＃39;);
}
$redis->auth(&＃39;&＃39;);
$redis->select(0);
// 投递任务
$data = [
    &＃39;to&＃39;      => [&＃39;***@qq.com&＃39; => &＃39;A name&＃39;],
    &＃39;body&＃39;    => &＃39;Here is the message itself&＃39;,
    &＃39;subject&＃39; => &＃39;The title content&＃39;,
];
$redis->lpush(&＃39;queue:email&＃39;, serialize($data));

通常异步开发中，投递完成后就会立即响应一个消息给用户，当然此时该任务并没有执行。

消费者开发

本例我们使用 MixPHP 的多进程开发工具 TaskExecutor 来完成这个需求，通常使用常驻进程来处理队列的消费，所以我们使用 TaskExecutor 的 TYPE_DAEMON 类型，MODE_PUSH 模式。

TaskExecutor 的 MODE_PUSH 模式有二种进程：

• 左进程：负责从消息队列取出任务数据，投放给中进程。

• 中进程：负责执行邮件发送任务。

PushCommand.php 代码如下：

 */
class PushCommand extends BaseCommand
{

    // 配置信息
    const HOST = &＃39;smtpdm.aliyun.com&＃39;;
    const PORT = 465;
    const SECURITY = &＃39;ssl&＃39;;
    const USERNAME = &＃39;****@email.***.com&＃39;;
    const PASSWORD = &＃39;****&＃39;;

    // 初始化事件
    public function onInitialize()
    {
        parent::onInitialize(); // TODO: Change the autogenerated stub
        // 获取程序名称
        $this->programName = Input::getCommandName();
        // 设置pidfile
        $this->pidFile = "/var/run/{$this->programName}.pid";
    }

    /**
     * 获取服务
     * @return TaskExecutor
     */
    public function getTaskService()
    {
        return create_object(
            [
                // 类路径
                &＃39;class&＃39;         => &＃39;mix\task\TaskExecutor&＃39;,
                // 服务名称
                &＃39;name&＃39;          => "mix-daemon: {$this->programName}",
                // 执行类型
                &＃39;type&＃39;          => \mix\task\TaskExecutor::TYPE_DAEMON,
                // 执行模式
                &＃39;mode&＃39;          => \mix\task\TaskExecutor::MODE_PUSH,
                // 左进程数
                &＃39;leftProcess&＃39;   => 1,
                // 中进程数
                &＃39;centerProcess&＃39; => 5,
                // 任务超时时间 (秒)
                &＃39;timeout&＃39;       => 5,
            ]
        );
    }

    // 启动
    public function actionStart()
    {
        // 预处理
        if (!parent::actionStart()) {
            return ExitCode::UNSPECIFIED_ERROR;
        }
        // 启动服务
        $service = $this->getTaskService();
        $service->on(&＃39;LeftStart&＃39;, [$this, &＃39;onLeftStart&＃39;]);
        $service->on(&＃39;CenterStart&＃39;, [$this, &＃39;onCenterStart&＃39;]);
        $service->start();
        // 返回退出码
        return ExitCode::OK;
    }

    // 左进程启动事件回调函数
    public function onLeftStart(LeftProcess $worker)
    {
        try {
            // 模型内使用长连接版本的数据库组件，这样组件会自动帮你维护连接不断线
            $queueModel = Redis::getInstance();
            // 保持任务执行状态，循环结束后当前进程会退出，主进程会重启一个新进程继续执行任务，这样做是为了避免长时间执行内存溢出
            for ($j = 0; $j <16000; $j++) {
                // 从消息队列中间件阻塞获取一条消息
                $data = $queueModel->brpop(&＃39;queue:email&＃39;, 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 出现 100%
            sleep(1);
            // 抛出错误
            throw $e;
        }
    }

    // 中进程启动事件回调函数
    public function onCenterStart(CenterProcess $worker)
    {
        // 保持任务执行状态，循环结束后当前进程会退出，主进程会重启一个新进程继续执行任务，这样做是为了避免长时间执行内存溢出
        for ($j = 0; $j <16000; $j++) {
            // 从进程消息队列中抢占一条消息
            $data = $worker->pop();
            if (empty($data)) {
                continue;
            }
            // 处理消息
            try {
                // 处理消息，比如：发送短信、发送邮件、微信推送
                var_dump($data);
                $ret = self::sendEmail($data);
                var_dump($ret);
            } catch (\Exception $e) {
                // 回退数据到消息队列
                $worker->rollback($data);
                // 休息一会，避免 CPU 出现 100%
                sleep(1);
                // 抛出错误
                throw $e;
            }
        }
    }

    // 发送邮件
    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[&＃39;subject&＃39;]))
            ->setFrom([self::USERNAME => &＃39;**网&＃39;])
            ->setTo($data[&＃39;to&＃39;])
            ->setBody($data[&＃39;body&＃39;]);
        // Send the message
        $result = $mailer->send($message);
        return $result;
    }

}

测试

[root@localhost bin]# ./mix-daemon push start
mix-daemon &＃39;push&＃39; start successed.

此时 shell 终端将打印：

官网：http://www.mixphp1.cn/

以上就是教你使用mixphp打造多进程异步邮件发送的详细内容，更多请关注 第一PHP社区 其它相关文章！