Utilisez mixphp pour créer un envoi d'e-mails asynchrone multi-processus

Cet article présente principalement l'utilisation de mixphp pour créer un envoi d'e-mails asynchrone multi-processus. Il a une certaine valeur de référence. Maintenant, je le partage avec vous. Les amis dans le besoin peuvent s'y référer

L'envoi d'e-mails est un. exigence très courante., Étant donné que l'opération d'envoi d'e-mails prend généralement du temps, nous utilisons généralement le traitement asynchrone pour améliorer l'expérience utilisateur, et le traitement asynchrone est généralement mis en œuvre à l'aide de files d'attente de messages.

En raison du manque de capacités de développement multi-processus, le framework MVC traditionnel utilise généralement le même script pour être exécuté plusieurs fois pour générer plusieurs processus. Mixphp encapsule TaskExecutor spécifiquement pour le développement multi-processus, et les utilisateurs peuvent développer. très simplement. Une application multi-processus entièrement fonctionnelle et hautement disponible.

Ce qui suit démontre le processus de développement d'un système d'envoi d'e-mails asynchrone, impliquant des points de connaissance :

• Asynchrone

• File d'attente de messages

• Processus multiples

• Processus démon

Comment utiliser la file d'attente de messages pour implémenter l'asynchrone

L'utilisation par PHP des files d'attente de messages est généralement implémentée à l'aide d'un middleware de messages couramment utilisé :

• redis

• rabbitmq

  .

• kafka

Cette fois, nous choisissons redis pour implémenter l'envoi d'e-mails asynchrone. Il existe un type de liste dans le type de données de redis, qui peut implémenter une file d'attente de messages. . Utilisez la commande suivante :

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

Conception d'architecture

Cet exemple utilise le framework MVC traditionnel pour répondre aux exigences d'envoi d'e-mails et le multi-processus MixPHP pour effectuer la tâche d'envoi.

Sélection de la bibliothèque d'envoi d'e-mails

Dans le passé, nous utilisions généralement la bibliothèque d'envoi d'e-mails fournie par le framework, ou téléchargeions des bibliothèques partagées par d'autres utilisateurs en ligne Après l'apparition de composer, https://. packagist.org/ Il existe de nombreuses bibliothèques de haute qualité sur Internet, il suffit de choisir la meilleure. Dans ce cas, nous choisissons Swiftmailer.

La tâche d'envoi étant exécutée par MixPHP, Swiftmailer est installé dans le projet MixPHP. Exécutez la commande suivante dans le répertoire racine du projet pour installer :

composer require swiftmailer/swiftmailer

Développement Producer

<.>Dans l'exigence d'envoi d'e-mails, le producteur fait référence à la partie qui effectue la tâche d'envoi. Cette partie est généralement une interface ou une page Web. Cette partie n'a pas nécessairement besoin d'être développée avec mixphp, CI et YII. tout est acceptable. Il suffit de l'ajouter à l'interface ou à la page Web. Il suffit de publier les informations sur la tâche dans la file d'attente des messages.

Ajoutez le code suivant au contrôleur du framework MVC traditionnel :

Habituellement, l'utilisation de redis dans le framework installera une bibliothèque de classes à utiliser. Cet exemple utilise du code natif pour une compréhension facile.

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

Habituellement, dans le développement asynchrone, un message sera répondu à l'utilisateur immédiatement une fois la livraison terminée. Bien entendu, la tâche n'est pas exécutée à ce moment-là.

Développement consommateur

Dans cet exemple, nous utilisons l'outil de développement multi-processus de MixPHP, TaskExecutor, pour remplir cette exigence. Les processus résidents sont généralement utilisés pour gérer la consommation de file d'attente, nous utilisons donc le type TYPE_DAEMON de TaskExecutor. MODE_PUSH.

Le mode MODE_PUSH de TaskExecutor comporte deux processus :

• Processus de gauche : responsable de la suppression des données de tâche de la file d'attente des messages et de leur mise dans le processus intermédiaire.

• Processus moyen : Responsable de l'exécution des tâches d'envoi d'e-mails.

Le code PushCommand.php est le suivant :

<?php

namespace apps\daemon\commands;

use mix\console\ExitCode;
use mix\facades\Input;
use mix\facades\Redis;
use mix\task\CenterProcess;
use mix\task\LeftProcess;
use mix\task\TaskExecutor;

/**
 * 推送模式范例
 * @author 刘健 <coder.liu@qq.com>
 */
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;
    }

}

Test

1. Démarrer le programme résident push dans le shell .

[root@localhost bin]# ./mix-daemon push start
mix-daemon 'push' start successed.

1. Appelez l'interface pour mettre les tâches dans la file d'attente des messages.

À ce moment, le terminal shell imprimera :

Reçu avec succès l'e-mail de test :

Ce qui précède représente l'intégralité du contenu de cet article. J'espère qu'il sera utile à l'étude de chacun. Pour plus de contenu connexe, veuillez faire attention au site Web PHP chinois !

Recommandations associées :

Activer l'extension shmop pour PHP afin d'implémenter la mémoire partagée

php implémente la fonction de communication du processus de mémoire partagée ( _shm )

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!