本文目录导读：
我来详细介绍PHP项目中CLI脚本和守护进程的开发与应用。
CLI脚本基础
1 基本结构
#!/usr/bin/env php
<?php
/**
* CLI脚本示例
*
* 使用方法: php script.php [options] [arguments]
*/
// 确保在CLI模式下运行
if (php_sapi_name() !== 'cli') {
die("This script must be run from command line.\n");
}
// 错误处理设置
error_reporting(E_ALL);
ini_set('display_errors', 1);
// 参数解析
$options = getopt('h::v::', ['help::', 'verbose::']);
$args = array_slice($argv, 1);
// 帮助信息
function showHelp() {
echo <<<HELP
用法: php script.php [选项] [参数]
选项:
-h, --help 显示帮助信息
-v, --verbose 显示详细输出
HELP;
exit(0);
}
if (isset($options['h']) || isset($options['help'])) {
showHelp();
}
// 主逻辑
$verbose = isset($options['v']) || isset($options['verbose']);
echo "Script started.\n";
if ($verbose) {
echo "Arguments: " . implode(', ', $args) . "\n";
}
2 命令行参数处理
<?php
/**
* 高级命令行参数处理
*/
class CommandLineParser {
private $arguments = [];
private $options = [];
private $commands = [];
public function __construct() {
$this->parseArgs();
}
private function parseArgs() {
global $argv;
for ($i = 1; $i < count($argv); $i++) {
$arg = $argv[$i];
// 选项处理（--option=value 或 --option value）
if (strpos($arg, '--') === 0) {
$option = substr($arg, 2);
if (strpos($option, '=') !== false) {
list($key, $value) = explode('=', $option, 2);
$this->options[$key] = $value;
} else {
if ($i + 1 < count($argv) && !strpos($argv[$i + 1], '-')) {
$this->options[$option] = $argv[++$i];
} else {
$this->options[$option] = true;
}
}
}
// 短选项处理（-v 或 -abc）
elseif (strpos($arg, '-') === 0) {
$flags = substr($arg, 1);
for ($j = 0; $j < strlen($flags); $j++) {
$this->options[$flags[$j]] = true;
}
}
// 命令处理
elseif (empty($this->commands)) {
$this->commands[] = $arg;
}
// 参数处理
else {
$this->arguments[] = $arg;
}
}
}
public function getOption($key, $default = null) {
return $this->options[$key] ?? $default;
}
public function getCommand($index = 0) {
return $this->commands[$index] ?? null;
}
public function getArgument($index = 0) {
return $this->arguments[$index] ?? null;
}
}
// 使用示例
$parser = new CommandLineParser();
$command = $parser->getCommand();
$verbose = $parser->getOption('verbose', false);
守护进程开发
1 基础守护进程框架
<?php
/**
* 基础守护进程类
*/
class Daemon {
private $pidFile;
private $logFile;
private $running = false;
private $callback;
public function __construct($pidFile = '/tmp/daemon.pid', $logFile = '/tmp/daemon.log') {
$this->pidFile = $pidFile;
$this->logFile = $logFile;
}
/**
* 设置主循环回调
*/
public function setCallback(callable $callback) {
$this->callback = $callback;
}
/**
* 启动守护进程
*/
public function start() {
if ($this->isRunning()) {
die("Daemon is already running.\n");
}
// 创建子进程
$pid = pcntl_fork();
if ($pid == -1) {
die("Fork failed.\n");
} elseif ($pid) {
// 父进程退出
exit(0);
}
// 子进程继续执行
// 创建新的会话
if (posix_setsid() == -1) {
die("Setsid failed.\n");
}
// 第二次fork
$pid = pcntl_fork();
if ($pid == -1) {
die("Second fork failed.\n");
} elseif ($pid) {
exit(0);
}
// 保存PID
file_put_contents($this->pidFile, posix_getpid());
// 关闭标准输入输出
fclose(STDIN);
fclose(STDOUT);
fclose(STDERR);
// 重定向到日志文件
$logFile = fopen($this->logFile, 'a');
if ($logFile) {
ob_start(function($buffer) use ($logFile) {
fwrite($logFile, $buffer);
return '';
});
}
$this->running = true;
$this->log("Daemon started.");
// 信号处理
pcntl_signal(SIGTERM, [$this, 'handleSignal']);
pcntl_signal(SIGINT, [$this, 'handleSignal']);
// 主循环
while ($this->running) {
pcntl_signal_dispatch();
if ($this->callback) {
call_user_func($this->callback);
}
sleep(1);
}
$this->cleanup();
}
/**
* 停止守护进程
*/
public function stop() {
if ($pid = $this->getPid()) {
posix_kill($pid, SIGTERM);
$this->log("Stopping daemon...");
}
}
/**
* 重启守护进程
*/
public function restart() {
$this->stop();
sleep(1);
$this->start();
}
/**
* 检查是否运行
*/
public function status() {
if ($this->isRunning()) {
echo "Daemon is running (PID: " . $this->getPid() . ")\n";
} else {
echo "Daemon is not running.\n";
}
}
/**
* 检查守护进程是否在运行
*/
private function isRunning() {
$pid = $this->getPid();
if ($pid && posix_kill($pid, 0)) {
return true;
}
if (file_exists($this->pidFile)) {
unlink($this->pidFile);
}
return false;
}
/**
* 获取PID
*/
private function getPid() {
if (file_exists($this->pidFile)) {
return (int) file_get_contents($this->pidFile);
}
return null;
}
/**
* 信号处理
*/
public function handleSignal($signo) {
switch ($signo) {
case SIGTERM:
case SIGINT:
$this->running = false;
break;
}
}
/**
* 日志记录
*/
private function log($message) {
$timestamp = date('Y-m-d H:i:s');
$logMessage = "[{$timestamp}] {$message}\n";
file_put_contents($this->logFile, $logMessage, FILE_APPEND);
}
/**
* 清理
*/
private function cleanup() {
if (file_exists($this->pidFile)) {
unlink($this->pidFile);
}
$this->log("Daemon stopped.");
}
}
2 使用示例
<?php
// 创建CLI入口
if (php_sapi_name() !== 'cli') {
die("This script must be run from command line.\n");
}
require_once 'Daemon.php';
$daemon = new Daemon('/tmp/myapp.pid', '/tmp/myapp.log');
// 设置主循环逻辑
$daemon->setCallback(function() {
// 执行任务
echo "Working...\n";
// 处理数据库连接
static $db = null;
if ($db === null) {
try {
$pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'password');
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$db = $pdo;
} catch (PDOException $e) {
// 记录数据库连接错误
return;
}
}
// 执行具体任务
try {
$stmt = $db->query("SELECT * FROM queue WHERE status = 'pending' LIMIT 1");
if ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
processJob($row);
}
} catch (PDOException $e) {
// 重连数据库
$db = null;
}
});
// 根据命令行参数执行
$command = $argv[1] ?? 'help';
switch ($command) {
case 'start':
$daemon->start();
break;
case 'stop':
$daemon->stop();
break;
case 'restart':
$daemon->restart();
break;
case 'status':
$daemon->status();
break;
default:
echo "用法: php daemon.php {start|stop|restart|status}\n";
exit(1);
}
高级守护进程特性
1 进程管理
<?php
/**
* 进程管理器
*/
class ProcessManager {
private $workers = [];
private $maxWorkers = 5;
private $tasks = [];
public function __construct($maxWorkers = 5) {
$this->maxWorkers = $maxWorkers;
}
/**
* 添加任务
*/
public function addTask($task) {
$this->tasks[] = $task;
$this->processTasks();
}
/**
* 处理任务队列
*/
private function processTasks() {
while (count($this->workers) < $this->maxWorkers && !empty($this->tasks)) {
$task = array_shift($this->tasks);
$this->spawnWorker($task);
}
}
/**
* 创建子进程
*/
private function spawnWorker($task) {
$pid = pcntl_fork();
if ($pid == -1) {
// Fork失败
return false;
} elseif ($pid) {
// 父进程
$this->workers[$pid] = $task;
return true;
} else {
// 子进程执行任务
try {
$result = call_user_func($task['callback'], $task['data'] ?? null);
// 将结果写入共享内存或管道
$this->writeResult($task['id'], $result);
} catch (Exception $e) {
$this->writeResult($task['id'], null, $e->getMessage());
}
exit(0);
}
}
/**
* 回收子进程
*/
public function reapChildren() {
while (($pid = pcntl_wait($status, WNOHANG)) > 0) {
if (isset($this->workers[$pid])) {
unset($this->workers[$pid]);
}
}
}
/**
* 写入结果到共享存储
*/
private function writeResult($taskId, $result, $error = null) {
// 使用文件或共享内存存储结果
$data = [
'task_id' => $taskId,
'result' => $result,
'error' => $error,
'time' => time()
];
file_put_contents("/tmp/task_result_{$taskId}", serialize($data));
}
/**
* 获取结果
*/
public function getResult($taskId) {
$file = "/tmp/task_result_{$taskId}";
if (file_exists($file)) {
$data = unserialize(file_get_contents($file));
unlink($file);
return $data;
}
return null;
}
}
2 信号处理和优雅关闭
<?php
/**
* 信号处理器
*/
class SignalHandler {
private $signals = [];
private $callbacks = [];
private $debug = false;
public function __construct($debug = false) {
$this->debug = $debug;
$this->registerDefaultSignals();
}
/**
* 注册默认信号处理
*/
private function registerDefaultSignals() {
$signals = [
SIGTERM => 'terminate',
SIGINT => 'interrupt',
SIGHUP => 'reload',
SIGUSR1 => 'user1',
SIGUSR2 => 'user2',
];
foreach ($signals as $signal => $name) {
$this->register($signal, function() use ($name) {
$this->handleSignal($name);
});
}
}
/**
* 注册信号处理器
*/
public function register($signal, callable $callback) {
if (!in_array($signal, $this->signals)) {
$this->signals[] = $signal;
}
$this->callbacks[$signal] = $callback;
pcntl_signal($signal, function($signo) use ($signal) {
if (isset($this->callbacks[$signal])) {
call_user_func($this->callbacks[$signal], $signo);
}
});
}
/**
* 处理信号
*/
private function handleSignal($name) {
if ($this->debug) {
echo "Received signal: {$name}\n";
}
switch ($name) {
case 'terminate':
case 'interrupt':
$this->gracefulShutdown();
break;
case 'reload':
$this->reloadConfiguration();
break;
default:
// 自定义信号处理
break;
}
}
/**
* 优雅关闭
*/
private function gracefulShutdown() {
if ($this->debug) {
echo "Starting graceful shutdown...\n";
}
// 停止接受新工作
// 完成当前工作
// 清理资源
// 关闭连接
exit(0);
}
/**
* 重新加载配置
*/
private function reloadConfiguration() {
if ($this->debug) {
echo "Reloading configuration...\n";
}
// 重新读取配置文件
// 更新运行时配置
// 不中断服务
}
/**
* 分发信号
*/
public function dispatch() {
pcntl_signal_dispatch();
}
}
实际应用示例
1 消息队列消费者
<?php
/**
* 消息队列消费者守护进程
*/
class QueueConsumerDaemon {
private $redis;
private $daemon;
private $config;
public function __construct($config) {
$this->config = $config;
$this->initRedis();
$this->initDaemon();
}
private function initRedis() {
$this->redis = new Redis();
$this->redis->connect(
$this->config['redis_host'] ?? '127.0.0.1',
$this->config['redis_port'] ?? 6379
);
}
private function initDaemon() {
$this->daemon = new Daemon(
$this->config['pid_file'] ?? '/tmp/queue_consumer.pid',
$this->config['log_file'] ?? '/tmp/queue_consumer.log'
);
$this->daemon->setCallback([$this, 'processQueue']);
}
public function processQueue() {
// 从队列中获取消息
$message = $this->redis->brPop($this->config['queue_name'], 1);
if ($message) {
$data = json_decode($message[1], true);
if ($data) {
try {
$this->processMessage($data);
} catch (Exception $e) {
// 记录错误并将消息放入死信队列
$this->redis->lPush(
$this->config['dead_letter_queue'] ?? 'dead_letter',
json_encode([
'original' => $data,
'error' => $e->getMessage(),
'time' => time()
])
);
}
}
}
}
private function processMessage($data) {
// 处理具体消息
switch ($data['type']) {
case 'email':
$this->sendEmail($data);
break;
case 'notification':
$this->sendNotification($data);
break;
case 'webhook':
$this->callWebhook($data);
break;
}
}
}
2 任务调度器
<?php
/**
* 定时任务调度器
*/
class TaskScheduler {
private $tasks = [];
private $running = true;
public function addTask($name, callable $callback, $interval, $options = []) {
$this->tasks[$name] = [
'callback' => $callback,
'interval' => $interval,
'last_run' => 0,
'enabled' => $options['enabled'] ?? true,
'timeout' => $options['timeout'] ?? 3600
];
}
public function run() {
while ($this->running) {
foreach ($this->tasks as $name => &$task) {
if (!$task['enabled']) {
continue;
}
$now = time();
if ($now - $task['last_run'] >= $task['interval']) {
$task['last_run'] = $now;
// 使用进程隔离执行任务
$pid = pcntl_fork();
if ($pid == -1) {
// Fork失败
continue;
} elseif ($pid) {
// 父进程，设置超时监控
$this->monitorTask($name, $pid, $task['timeout']);
} else {
// 子进程执行任务
try {
call_user_func($task['callback']);
} catch (Exception $e) {
// 记录任务执行失败
$this->logTaskError($name, $e);
}
exit(0);
}
}
}
// 回收子进程
pcntl_wait($status, WNOHANG);
sleep(1);
}
}
private function monitorTask($name, $pid, $timeout) {
// 创建超时监控
$start = time();
while (time() - $start < $timeout) {
if (!posix_kill($pid, 0)) {
// 进程已结束
return;
}
usleep(100000); // 100ms
}
// 超时，强制终止
posix_kill($pid, SIGKILL);
$this->logTaskTimeout($name);
}
}
运维与监控
1 健康检查
<?php
/**
* 健康检查API
*/
class HealthCheck {
private $checks = [];
public function addCheck($name, callable $check) {
$this->checks[$name] = $check;
}
public function runCheck() {
$status = true;
$results = [];
foreach ($this->checks as $name => $check) {
try {
$result = call_user_func($check);
$results[$name] = [
'status' => $result ? 'ok' : 'failed',
'message' => $result ? 'OK' : 'Failed'
];
if (!$result) {
$status = false;
}
} catch (Exception $e) {
$results[$name] = [
'status' => 'error',
'message' => $e->getMessage()
];
$status = false;
}
}
return [
'status' => $status ? 'healthy' : 'unhealthy',
'timestamp' => time(),
'checks' => $results
];
}
}
2 systemd服务配置
# /etc/systemd/system/php-daemon.service [Unit] Description=PHP Daemon Service After=network.target [Service] Type=simple User=www-data Group=www-data WorkingDirectory=/var/www/app ExecStart=/usr/bin/php /var/www/app/daemon.php start ExecStop=/usr/bin/php /var/www/app/daemon.php stop ExecReload=/usr/bin/php /var/www/app/daemon.php restart Restart=always RestartSec=5 TimeoutStopSec=30 KillSignal=SIGTERM [Install] WantedBy=multi-user.target
最佳实践
1 错误处理
<?php
/**
* 全局错误处理
*/
class ErrorHandler {
public static function register() {
set_error_handler([self::class, 'handleError']);
set_exception_handler([self::class, 'handleException']);
register_shutdown_function([self::class, 'handleShutdown']);
}
public static function handleError($severity, $message, $file, $line) {
if (!(error_reporting() & $severity)) {
return;
}
throw new ErrorException($message, 0, $severity, $file, $line);
}
public static function handleException($exception) {
// 记录异常到日志
error_log("Uncaught Exception: " . $exception->getMessage() .
" in " . $exception->getFile() .
":" . $exception->getLine());
// 发送告警
// ...
// 优雅关闭
if (function_exists('posix_kill')) {
posix_kill(posix_getpid(), SIGTERM);
}
}
public static function handleShutdown() {
$error = error_get_last();
if ($error && in_array($error['type'], [E_ERROR, E_PARSE, E_CORE_ERROR, E_COMPILE_ERROR])) {
// 记录致命错误
error_log("Fatal Error: " . $error['message'] .
" in " . $error['file'] .
":" . $error['line']);
}
}
}
2 资源管理
<?php
/**
* 资源管理器
*/
class ResourceManager {
private static $instances = [];
private $connections = [];
private $files = [];
/**
* 注册资源
*/
public function register($name, $resource, $type = 'connection') {
$this->resources[$name] = [
'resource' => $resource,
'type' => $type,
'created_at' => time()
];
}
/**
* 释放所有资源
*/
public function releaseAll() {
foreach ($this->resources as $name => $info) {
switch ($info['type']) {
case 'connection':
if ($info['resource'] instanceof PDO) {
$info['resource'] = null;
}
break;
case 'file':
if (is_resource($info['resource'])) {
fclose($info['resource']);
}
break;
case 'stream':
if (is_resource($info['resource'])) {
stream_socket_shutdown($info['resource'], STREAM_SHUT_RDWR);
}
break;
}
}
$this->resources = [];
}
/**
* 获取资源使用统计
*/
public function getStats() {
$stats = [
'total_resources' => count($this->resources),
'memory_usage' => memory_get_usage(true),
'peak_memory' => memory_get_peak_usage(true)
];
return $stats;
}
}
性能优化
1 内存优化
<?php
/**
* 内存优化示例
*/
class MemoryOptimizer {
/**
* 批量处理，避免内存溢出
*/
public function processLargeDataset($callback, $batchSize = 1000) {
$offset = 0;
while (true) {
$batch = $this->getDataBatch($offset, $batchSize);
if (empty($batch)) {
break;
}
foreach ($batch as $item) {
call_user_func($callback, $item);
}
// 强制垃圾回收
gc_collect_cycles();
$offset += $batchSize;
}
}
/**
* 使用生成器处理大数据
*/
public function processWithGenerator($data) {
foreach ($data as $item) {
yield $this->processItem($item);
// 每处理1000个项目清理一次
if (++$count % 1000 === 0) {
gc_collect_cycles();
}
}
}
}
CLI脚本和守护进程开发需要注意内存管理、错误处理、信号处理等关键方面，选择适合的场景使用正确的方式来实现,可以提高项目的稳定性和可维护性。
