WorkerMan 与多进程深度解析
WorkerMan 的核心架构
WorkerMan 是一个基于 PHP 的高性能 socket 框架，其核心设计思想就是多进程架构：
┌─────────────────────────────────────────┐
│ Master 进程 │
│ (管理进程，负责监控和信号分发) │
└─────────────────────────────────────────┘
│
┌───────────────┼───────────────┐
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Worker1 │ │ Worker2 │ │ Worker3 │ ...
│ (子进程) │ │ (子进程) │ │ (子进程) │
│ 监听端口 │ │ 监听端口 │ │ 监听端口 │
└─────────┘ └─────────┘ └─────────┘多进程工作模式
1 进程模型类型
use Workerman\Worker;
// 1. 单进程模式（不推荐生产环境）
$worker = new Worker('tcp://0.0.0.0:1234');
$worker->count = 1;
// 2. 多进程模式（推荐）
$http_worker = new Worker('http://0.0.0.0:8080');
$http_worker->count = 4; // 启动4个进程
// 3. 多端口、多协议
$tcp_worker = new Worker('tcp://0.0.0.0:8081');
$tcp_worker->count = 2;
$ws_worker = new Worker('websocket://0.0.0.0:8082');
$ws_worker->count = 3;
2 进程间通信（IPC）
// 使用 Channel 进行进程间通信
use Workerman\Channel\Client;
use Workerman\Channel\Server;
// 启动 Channel 服务
$channel_server = new Server();
// Worker 进程
$worker = new Worker();
$worker->onWorkerStart = function() {
// 订阅消息
Client::on('message', function($event_data) {
echo "Received: " . $event_data . PHP_EOL;
});
};
// 发送进程间消息
$worker2 = new Worker();
$worker2->onWorkerStart = function() {
Client::publish('message', 'Hello from another process');
};
多进程关键特性
1 进程隔离与资源共享
// 每个进程独立的内存空间
$worker = new Worker('http://0.0.0.0:8080');
$worker->count = 4;
// 全局变量在不同进程间不共享
$global_counter = 0;
$worker->onMessage = function($connection, $data) {
global $global_counter;
$global_counter++; // 每个进程独立计数，不会互相影响
// 需要共享数据时使用 Channel 或外部存储
$connection->send("Process: " . posix_getpid());
};
2 连接分配机制
// 默认使用轮询分配连接
$worker = new Worker('tcp://0.0.0.0:8888');
$worker->count = 8;
// 设置连接分配策略
$worker->reusePort = true; // Linux下启用端口复用
// 自定义连接分配
$worker->onConnect = function($connection) {
// 可以根据IP、连接数等策略分配
$ip = $connection->getRemoteIp();
// 自定义分配逻辑
};
实际应用场景
1 高并发 WebSocket 服务
class ChatServer {
public function __construct() {
$worker = new Worker('websocket://0.0.0.0:8080');
$worker->count = 8;
$worker->onConnect = function($connection) {
$connection->uid = uniqid();
};
$worker->onMessage = function($connection, $data) {
// 广播消息到所有连接
foreach ($connection->worker->connections as $conn) {
$conn->send($data);
}
};
$worker->onClose = function($connection) {
echo "Connection closed: {$connection->uid}\n";
};
}
}
2 任务分发系统
// 主进程分发任务
class TaskDispatcher {
private $workers = [];
public function init() {
// 创建多个工作进程
for ($i = 0; $i < 4; $i++) {
$worker = new Worker();
$worker->name = "TaskWorker-{$i}";
$worker->count = 1;
$worker->onWorkerStart = function($worker) {
// 每个进程处理自己的任务队列
while (true) {
$task = $this->getTaskFromQueue($worker->id);
if ($task) {
$this->processTask($task);
}
sleep(1);
}
};
$this->workers[] = $worker;
}
}
private function getTaskFromQueue($workerId) {
// 从Redis或其他队列获取任务
// 使用进程ID进行负载均衡
return [
'id' => $workerId,
'data' => "Task for process {$workerId}"
];
}
}
性能优化策略
1 进程数优化
class OptimizedServer {
public function start() {
$worker = new Worker('http://0.0.0.0:8080');
// CPU密集型：进程数 = CPU核心数
$cpuCores = shell_exec('nproc');
$worker->count = (int)$cpuCores;
// IO密集型：进程数 = CPU核心数 * 2
$worker->count = (int)$cpuCores * 2;
$worker->onMessage = function($connection, $data) {
// 处理请求
$connection->send('response');
};
}
}
2 内存优化
// 限定每个进程的内存使用
$worker = new Worker();
$worker->count = 4;
$worker->maxRequest = 10000; // 处理10000个请求后重启进程
$worker->onMessage = function($connection, $data) {
if (memory_get_usage() > 100 * 1024 * 1024) { // 超过100MB
// 主动退出进程，主进程会重新创建
Worker::stop();
}
};
常见问题与解决方案
1 进程死锁问题
class DeadlockPrevention {
public static function start() {
$worker = new Worker();
$worker->count = 4;
// 设置进程超时（毫秒）
$worker->stopTimeout = 5;
// 优雅重启
$worker->reloadable = true;
// 心跳检测
$worker->onConnect = function($connection) {
$connection->heartbeatTime = time();
};
$worker->onMessage = function($connection, $data) {
// 处理业务逻辑
if (time() - $connection->heartbeatTime > 30) {
$connection->close('Heartbeat timeout');
}
};
}
}
2 数据一致性
// 使用 Redis 共享状态
class SharedState {
private $redis;
public function __construct() {
$this->redis = new \Redis();
$this->redis->connect('127.0.0.1', 6379);
}
public function atomicUpdate($key, $callback) {
// 使用 Redis 乐观锁
$this->redis->watch($key);
$value = $this->redis->get($key);
$newValue = $callback($value);
$this->redis->multi();
$this->redis->set($key, $newValue);
$this->redis->exec();
}
}
监控与管理
// 进程状态监控
class ProcessMonitor {
public function __construct() {
$monitor = new Worker();
$monitor->name = 'Monitor';
$monitor->onWorkerStart = function() {
while (true) {
// 检查所有工作进程状态
foreach (Worker::getAllWorkers() as $worker) {
$status = [
'pid' => $worker->id,
'connections' => count($worker->connections),
'memory' => memory_get_usage(true),
'start_time' => $worker->startTime
];
// 记录到日志或数据库
$this->logWorkerStatus($status);
}
sleep(60); // 每分钟检查一次
}
};
}
private function logWorkerStatus($status) {
file_put_contents(
'/var/log/workerman/status.log',
json_encode($status) . PHP_EOL,
FILE_APPEND
);
}
}
部署建议
# 1. 设置合适的工作进程数 php worker.php start -d --process=auto # 2. 使用Supervisor管理进程组 [program:workerman] command=php /path/to/worker.php start -d process_name=%(program_name)s_%(process_num)02d numprocs=4 autostart=true autorestart=true user=www-data stdout_logfile=/var/log/workerman/out.log stderr_logfile=/var/log/workerman/err.log # 3. 平滑重启 kill -USR1 $(cat /var/run/workerman.pid)
WorkerMan 的多进程架构让 PHP 能够实现高性能的网络服务，通过合理配置和优化，可以轻松应对高并发场景，理解进程模型、掌握 IPC 通信、做好资源管理和监控，是充分发挥 WorkerMan 性能的关键。
