该服务依赖 Swoole 扩展,Windows下无法执行。

多进程任务服务

PHP 原生是不支持多进程的,多进程任务服务是基于 Swoole 开发,MixPHP 已经建立好了进程模型,参考范例代码即可非常容易的编写出多进程任务处理,充分利用多核性能,处理大量数据。

调用
mix\server\TaskServer \Mix::app()->createObject('name')

多进程队列服务不是组件,且非整个框架内频繁使用,所以应该定义在 objects 字段内使用 \Mix::app()->createObject 方法实例化。

多进程队列服务仍然属于守护进程,所以在 daemon 模块内开发。

使用场景

如:消息队列(MQ)消费处理,消息推送,数据采集等。

优点

  • 平滑重启:当 kill 主进程时,子进程处理完工作再退出,不丢失数据。
  • 高容错:子进程异常奔溃时,主进程将重建子进程。
  • 高性能:多进程运行,充分利用多个CPU并行计算,性能强劲。
  • 使用灵活:工作进程使用生产者消费者模型,生产者/消费者的数量都可自定义。

进程模型

范例

配置文件

// 对象配置
'objects'             => [

    // QueueServer
    'server' => [
        // 类路径
        'class'        => 'mix\server\TaskServer',
        // 左进程数
        'leftProcess'  => 1,
        // 右进程数
        'rightProcess' => 3,
    ],

],

控制器代码

  • 代码中 $msg = $queueModel->pop(); 应该使用堵塞方法从消息队列取出数据,比如:redis 的 brpop 方法,进程堵塞后将释放CPU,如果不堵塞,持续执行的循环会导致CPU 100%。

  • 代码中 $msg = $worker->pop(); 执行时,如果没有数据会堵塞进程,堵塞后将释放CPU,不会占用性能。

  • 代码中 $worker->checkMaster(); 必须要是 while 循环体内的第一行,它实现了主进程被 kill,子进程处理完工作再退出。

<?php

namespace console\daemon\command;

use mix\console\Controller;

/**
 * 这是一个多进程守护进程的范例
 * 进程模型为:生产者消费者模型
 * 你可以自由选择是左进程当生产者还是右进程当生产者,本范例是左进程当生产者
 * @author 刘健 <coder.liu@qq.com>
 */
class MultiController extends Controller
{

    // 是否后台运行
    protected $d = false;

    /**
     * 获取服务
     * @return \mix\server\TaskServer
     */
    public function getServer()
    {
        return \Mix::app()->createObject('taskServer');
    }

    // 启动守护进程
    public function actionStart()
    {
        // 蜕变为守护进程
        if ($this->d) {
            self::daemon();
        }
        // 启动服务
        $server       = $this->getServer();
        $server->name = $this->getControllerName();
        $server->on('LeftStart', [$this, 'onLeftStart']);
        $server->on('RightStart', [$this, 'onRightStart']);
        $server->start();
    }

    // 左进程启动事件回调函数
    public function onLeftStart(\mix\process\TaskProcess $worker)
    {
        // 模型内使用长连接版本的数据库组件,这样组件会自动帮你维护连接不断线
        $queueModel = new \console\common\model\QueueModel();
        // 循环执行任务
        for ($j = 0; $j < 16000; $j++) {
            $worker->checkMaster();
            // 从队列取出一条消息
            $msg = $queueModel->pop();
            // 将消息推送给消费者进程处理
            $worker->push($msg);
        }
    }

    // 右进程启动事件回调函数
    public function onRightStart(\mix\process\TaskProcess $worker, $index)
    {
        // 循环执行任务
        for ($j = 0; $j < 16000; $j++) {
            $worker->checkMaster();
            // 从队列中抢占一条消息
            $msg = $worker->pop();
            if (!empty($msg)) {
                // 处理消息,比如:发送短信、发送邮件、微信推送
                // ...
            }
        }
    }

}

启动与查看

在命令行使用以下命令启动。

mix-daemon multi/start

启动后使用如下 Linux 命令管理进程。

// 查找进程
ps -ef | grep multi

// 结束主进程
kill <PID>