当前位置:首页 > PHP教程 > php高级应用 > 列表

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

发布:smiling 来源: PHP粉丝网  添加日期:2022-07-26 12:40:44 浏览: 评论:0 

注意:这个是 MixPHP V1 的范例

邮件发送是很常见的需求,由于发送邮件的操作一般是比较耗时的,所以我们一般采用异步处理来提升用户体验,而异步通常我们使用消息队列来实现。

传统 MVC 框架由于缺少多进程开发能力,通常是采用同一个脚本执行多次,产生多个进程的方式,mixphp 封装了 TaskExecutor 专用于多进程开发,用户能非常简单的开发出功能完善的高可用多进程应用。

下面演示一个异步邮件发送系统的开发过程,涉及知识点:

异步

消息队列

多进程

守护进程

如何使用消息队列实现异步

PHP 使用消息队列通常是使用中间件来实现,常用的消息中间件有:

redis

rabbitmq

kafka

本次我们选用 redis 来实现异步邮件发送,redis 的数据类型中有一个 list 类型,可实现消息队列,使用以下命令:

  1. // 入列 
  2.  
  3. $redis->lpush($key, $data); 
  4.  
  5. // 出列 
  6.  
  7. $data = $redis->rpop($key); 
  8.  
  9. // 阻塞出列 
  10.  
  11. $data = $redis->brpop($key, 10); 

架构设计

本实例由传统 MVC 框架投递邮件发送需求,MixPHP 多进程执行发送任务。

邮件发送库选型

以往我们通常使用框架提供的邮件发送库,或者网上下载别的用户分享的库,composer 出现后,https://packagist.org/ 上有大量优质的库,我们只需选择一个最好的即可,本例选择 swiftmailer。

由于发送任务是由 MixPHP 执行,所以 swiftmailer 是安装在 MixPHP 项目中,在项目根目录中执行以下命令安装:

composer require swiftmailer/swiftmailer

生产者开发

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

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

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

  1. // 连接 
  2.  
  3. $redis = new \Redis(); 
  4.  
  5. if (!$redis->connect('127.0.0.1', 6379)) { 
  6.  
  7.     throw new \Exception('Redis Connect Failure'); 
  8.  
  9. } 
  10.  
  11. $redis->auth(''); 
  12.  
  13. $redis->select(0); 
  14.  
  15. // 投递任务 
  16.  
  17. $data = [ 
  18.  
  19.     'to'      => ['***@qq.com' => 'A name'], 
  20.  
  21.     'body'    => 'Here is the message itself', 
  22.  
  23.     'subject' => 'The title content', 
  24.  
  25. ]; 
  26.  
  27. $redis->lpush('queue:email', serialize($data)); 

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

消费者开发

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

TaskExecutor 的 MODE_PUSH 模式有二种进程:

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

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

PushCommand.php 代码如下:

  1. <?php 
  2.  
  3. namespace apps\daemon\commands; 
  4.  
  5. use mix\console\ExitCode; 
  6.  
  7. use mix\facades\Input; 
  8.  
  9. use mix\facades\Redis; 
  10.  
  11. use mix\task\CenterProcess; 
  12.  
  13. use mix\task\LeftProcess; 
  14.  
  15. use mix\task\TaskExecutor; 
  16.  
  17.  
  18.  
  19. /** 
  20.  
  21.  * 推送模式范例 
  22.  
  23.  * @author 刘健 <coder.liu@qq.com> 
  24.  
  25.  */ 
  26.  
  27. class PushCommand extends BaseCommand 
  28.  
  29. { 
  30.  
  31.  
  32.  
  33.     // 配置信息 
  34.  
  35.     const HOST = 'smtpdm.aliyun.com'; 
  36.  
  37.     const PORT = 465; 
  38.  
  39.     const SECURITY = 'ssl'; 
  40.  
  41.     const USERNAME = '****@email.***.com'; 
  42.  
  43.     const PASSWORD = '****'; 
  44.  
  45.  
  46.  
  47.     // 初始化事件 
  48.  
  49.     public function onInitialize() 
  50.  
  51.     { 
  52.  
  53.         parent::onInitialize(); // TODO: Change the autogenerated stub 
  54.  
  55.         // 获取程序名称 
  56.  
  57.         $this->programName = Input::getCommandName(); 
  58.  
  59.         // 设置pidfile 
  60.  
  61.         $this->pidFile = "/var/run/{$this->programName}.pid"; 
  62.  
  63.     } 
  64.  
  65.  
  66.  
  67.     /** 
  68.  
  69.      * 获取服务 
  70.  
  71.      * @return TaskExecutor 
  72.  
  73.      */ 
  74.  
  75.     public function getTaskService() 
  76.  
  77.     { 
  78.  
  79.         return create_object( 
  80.  
  81.             [ 
  82.  
  83.                 // 类路径 
  84.  
  85.                 'class'         => 'mix\task\TaskExecutor', 
  86.  
  87.                 // 服务名称 
  88.  
  89.                 'name'          => "mix-daemon: {$this->programName}", 
  90.  
  91.                 // 执行类型 
  92.  
  93.                 'type'          => \mix\task\TaskExecutor::TYPE_DAEMON, 
  94.  
  95.                 // 执行模式 
  96.  
  97.                 'mode'          => \mix\task\TaskExecutor::MODE_PUSH, 
  98.  
  99.                 // 左进程数 
  100.  
  101.                 'leftProcess'   => 1, 
  102.  
  103.                 // 中进程数 
  104.  
  105.                 'centerProcess' => 5, 
  106.  
  107.                 // 任务超时时间 (秒) 
  108.  
  109.                 'timeout'       => 5, 
  110.  
  111.             ] 
  112.  
  113.         ); 
  114.  
  115.     } 
  116.  
  117.  
  118.  
  119.     // 启动 
  120.  
  121.     public function actionStart() 
  122.  
  123.     { 
  124.  
  125.         // 预处理 
  126.  
  127.         if (!parent::actionStart()) { 
  128.  
  129.             return ExitCode::UNSPECIFIED_ERROR; 
  130.  
  131.         } 
  132.  
  133.         // 启动服务 
  134.  
  135.         $service = $this->getTaskService(); 
  136.  
  137.         $service->on('LeftStart', [$this, 'onLeftStart']); 
  138.  
  139.         $service->on('CenterStart', [$this, 'onCenterStart']); 
  140.  
  141.         $service->start(); 
  142.  
  143.         // 返回退出码 
  144.  
  145.         return ExitCode::OK; 
  146.  
  147.     } 
  148.  
  149.  
  150.  
  151.     // 左进程启动事件回调函数 
  152.  
  153.     public function onLeftStart(LeftProcess $worker) 
  154.  
  155.     { 
  156.  
  157.         try { 
  158.  
  159.             // 模型内使用长连接版本的数据库组件,这样组件会自动帮你维护连接不断线 
  160.  
  161.             $queueModel = Redis::getInstance(); 
  162.  
  163.             // 保持任务执行状态,循环结束后当前进程会退出,主进程会重启一个新进程继续执行任务,这样做是为了避免长时间执行内存溢出 
  164.  
  165.             for ($j = 0; $j < 16000; $j++) { 
  166.  
  167.                 // 从消息队列中间件阻塞获取一条消息 
  168.  
  169.                 $data = $queueModel->brpop('queue:email', 10); 
  170.  
  171.                 if (emptyempty($data)) { 
  172.  
  173.                     continue; 
  174.  
  175.                 } 
  176.  
  177.                 list(, $data) = $data; 
  178.  
  179.                 // 将消息推送给中进程去处理,push有长度限制 (https://wiki.swoole.com/wiki/page/290.html) 
  180.  
  181.                 $worker->push($data, false); 
  182.  
  183.             } 
  184.  
  185.         } catch (\Exception $e) { 
  186.  
  187.             // 休息一会,避免 CPU 出现 100% 
  188.  
  189.             sleep(1); 
  190.  
  191.             // 抛出错误 
  192.  
  193.             throw $e; 
  194.  
  195.         } 
  196.  
  197.     } 
  198.  
  199.  
  200.  
  201.     // 中进程启动事件回调函数 
  202.  
  203.     public function onCenterStart(CenterProcess $worker) 
  204.  
  205.     { 
  206.  
  207.         // 保持任务执行状态,循环结束后当前进程会退出,主进程会重启一个新进程继续执行任务,这样做是为了避免长时间执行内存溢出 
  208.  
  209.         for ($j = 0; $j < 16000; $j++) { 
  210.  
  211.             // 从进程消息队列中抢占一条消息 
  212.  
  213.             $data = $worker->pop(); 
  214.  
  215.             if (emptyempty($data)) { 
  216.  
  217.                 continue; 
  218.  
  219.             } 
  220.  
  221.             // 处理消息 
  222.  
  223.             try { 
  224.  
  225.                 // 处理消息,比如:发送短信、发送邮件、微信推送 
  226.  
  227.                 var_dump($data); 
  228.  
  229.                 $ret = self::sendEmail($data); 
  230.  
  231.                 var_dump($ret); 
  232.  
  233.             } catch (\Exception $e) { 
  234.  
  235.                 // 回退数据到消息队列 
  236.  
  237.                 $worker->rollback($data); 
  238.  
  239.                 // 休息一会,避免 CPU 出现 100% 
  240.  
  241.                 sleep(1); 
  242.  
  243.                 // 抛出错误 
  244.  
  245.                 throw $e; 
  246.  
  247.             } 
  248.  
  249.         } 
  250.  
  251.     } 
  252.  
  253.  
  254.  
  255.     // 发送邮件 
  256.  
  257.     public static function sendEmail($data) 
  258.  
  259.     { 
  260.  
  261.         // Create the Transport 
  262.  
  263.         $transport = (new \Swift_SmtpTransport(self::HOST, self::PORT, self::SECURITY)) 
  264.  
  265.             ->setUsername(self::USERNAME) 
  266.  
  267.             ->setPassword(self::PASSWORD); 
  268.  
  269.         // Create the Mailer using your created Transport 
  270.  
  271.         $mailer = new \Swift_Mailer($transport); 
  272.  
  273.         // Create a message 
  274.  
  275.         $message = (new \Swift_Message($data['subject'])) 
  276.  
  277.             ->setFrom([self::USERNAME => '**网']) 
  278.  
  279.             ->setTo($data['to']) 
  280.  
  281.             ->setBody($data['body']); 
  282.  
  283.         // Send the message 
  284.  
  285.         $result = $mailer->send($message); 
  286.  
  287.         return $result; 
  288.  
  289.     } 
  290.  
  291.  
  292.  
  293. } 

测试

1.在 shell 中启动 push 常驻程序。

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

mix-daemon 'push' start successed.

1.调用接口往消息队列投放任务。

此时 shell 终端将打印:

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

成功收到测试邮件:

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

MixPHP

GitHub: https://github.com/mix-php/mix

官网:http://www.mixphp.cn/

Tags: mixphp多进程 php异步邮件发送

分享到: