PHP Swoole 守护进程开发实例：后台常驻与线程池处理

<?php
class ExampleWorker extends Worker {
    // 数据库连接句柄，供子任务复用
    protected static $dbh;

    public function __construct() {}

    public function run(){
        // 在子线程中初始化数据库连接
        $dbhost = '192.168.2.1';
        $dbport = 3306;
        $dbuser = 'www';
        $dbpass = 'qwer123';
        $dbname = 'example';

        self::$dbh = new PDO("mysql:host=$dbhost;port=$dbport;dbname=$dbname", $dbuser, $dbpass, array(
            PDO::MYSQL_ATTR_COMPRESS => true,
            PDO::ATTR_PERSISTENT => true
        ));
    }

    protected function getInstance(){
        return self::$dbh;
    }
}

// 可收集的任务类，用于 Pool::collect 回收
/* the collectable class implements machinery for Pool::collect */
class Fee extends Stackable {
    public function __construct($msg) {
        // 解析传入的消息
        $trades = explode(",", $msg);
        $this->data = $trades;
        print_r($trades);
    }

    public function run() {
        try {
            $dbh = $this->worker->getInstance();
            
            // 执行插入操作
            $insert = "INSERT INTO fee(ticket, login, volume, `status`) VALUES(:ticket, :login, :volume,'N')";
            $sth = $dbh->prepare($insert);
            $sth->bindValue(':ticket', $this->data[0]);
            $sth->bindValue(':login', $this->data[1]);
            $sth->bindValue(':volume', $this->data[2]);
            $sth->execute();
            $sth = null;
            
            /* ...... */
            
            // 更新状态
            $update = "UPDATE fee SET `status` = 'Y' WHERE ticket = :ticket and `status` = 'N'";
            $sth = $dbh->prepare($update);
            $sth->bindValue(':ticket', $this->data[0]);
            $sth->execute();
        } catch(PDOException $e) {
            // 错误日志记录
            $error = sprintf("%s,%s\n", $mobile, $id );
            file_put_contents("mobile_error.log", $error, FILE_APPEND);
        }
    }
}

class Example {
    // 配置项
    const LISTEN = "tcp://192.168.2.15:5555";
    const MAXCONN = 100;
    const pidfile = __CLASS__;
    const uid = 80;
    const gid = 80;
    
    protected $pool = NULL;
    protected $zmq = NULL;

    public function __construct() {
        $this->pidfile = '/var/run/'.self::pidfile.'.pid';
    }

    private function daemon(){
        // 检查 PID 文件，防止重复启动
        if (file_exists($this->pidfile)) {
            echo "The file $this->pidfile exists.\n";
            exit();
        }
        
        $pid = pcntl_fork();
        if ($pid == -1) {
            die('could not fork');
        } else if ($pid) {
            // 父进程退出
            exit($pid);
        } else {
            // 子进程进入后台
            file_put_contents($this->pidfile, getmypid());
            // 降低权限至指定用户
            posix_setuid(self::uid);
            posix_setgid(self::gid);
            return(getmypid());
        }
    }

    private function start(){
        $pid = $this->daemon();
        // 创建线程池
        $this->pool = new Pool(self::MAXCONN, \ExampleWorker::class, []);
        // 绑定 ZeroMQ Socket
        $this->zmq = new ZMQSocket(new ZMQContext(), ZMQ::SOCKET_REP);
        $this->zmq->bind(self::LISTEN);
        
        // 循环接收并分发任务
        while ($message = $this->zmq->recv()) {
            $this->pool->submit(new Fee($message));
            $this->zmq->send('TRUE'); 
        }
        $this->pool->shutdown(); 
    }

    private function stop(){
        if (file_exists($this->pidfile)) {
            $pid = file_get_contents($this->pidfile);
            posix_kill($pid, 9); 
            unlink($this->pidfile);
        }
    }

    private function help($proc){
        printf("%s start | stop | help \n", $proc);
    }

    public function main($argv){
        if(count($argv) < 2){
            printf("please input help parameter\n");
            exit();
        }
        if($argv[1] === 'stop'){
            $this->stop();
        }else if($argv[1] === 'start'){
            $this->start();
        }else{
            $this->help($argv[0]);
        }
    }
}

$cgse = new Example();
$cgse->main($argv);

private function daemon(){
    // 1. 防重入检查
    if (file_exists($this->pidfile)) {
     echo "The file $this->pidfile exists.\n";
     exit();
    }
    
    $pid = pcntl_fork();
    if ($pid == -1) {
      die('could not fork');
    } else if ($pid) {
      // 父进程退出
      exit($pid);
    } else {
      // 2. 子进程执行
      // 将当前子进程的 PID 写入文件
      file_put_contents($this->pidfile, getmypid());
      // 3. 权限降级：从 root 切换到普通用户 (如 www)
      posix_setuid(self::uid);
      posix_setgid(self::gid);
      return(getmypid());
    }
}

private function stop(){
    if (file_exists($this->pidfile)) {
        $pid = file_get_contents($this->pidfile);
        // 强制杀死进程
        posix_kill($pid, 9); 
        unlink($this->pidfile);
    }
}