PHP小码农

Appearance

共享内存与信号量 ​

概述 ​

共享内存（Shared Memory）和信号量（Semaphore）是操作系统级别的进程间通信（IPC）机制。共享内存允许多个进程访问同一块物理内存，实现高速数据交换；信号量则用于协调多个进程对共享资源的访问，防止竞争条件。PHP 通过 shmop、SysvSemaphore、SysvSharedMemory 和 PHP 8.0+ 的 SharedMemory 扩展提供对这些机制的支持。

前置知识

阅读本节前，建议先了解：信号处理。共享内存和信号量主要用于多进程通信场景。

基础概念 ​

进程间通信（IPC） ​

由于 PHP 的每个请求通常运行在独立的进程中，进程间通信的需求在以下场景中尤为常见：

  1. 队列 Worker 通信：多个 worker 进程共享任务队列状态
  2. 缓存共享：跨进程共享计算结果（如 OPcache 的实现）
  3. 限流计数器：跨进程的全局计数器
  4. 进程同步：协调多个 worker 的执行顺序

共享内存 vs 其他 IPC 方式 ​

方式速度数据量复杂度适用场景
共享内存最快大高高频数据交换
消息队列快中中任务分发
管道快小低简单数据流
Socket中大高跨机器通信
文件慢大低持久化数据

详细说明 ​

shmop 共享内存操作 ​

基本读写 ​

php
<?php
declare(strict_types=1);

/**
 * shmop 共享内存封装
 */
class ShmopManager
{
    private int $shmId;
    private int $size;

    /**
     * 创建或打开共享内存
     */
    public function __construct(string $key, int $size = 1024)
    {
        $this->size = $size;
        $shmKey = ftok(__FILE__, (int) $key);

        // 创建共享内存段
        $this->shmId = shmop_open(
            $shmKey,
            'c', // 创建（如存在则打开）
            0644, // 权限
            $size
        );

        if ($this->shmId === false) {
            // 尝试打开已存在的
            $this->shmId = shmop_open($shmKey, 'w', 0, 0);
        }

        if ($this->shmId === false) {
            throw new RuntimeException('无法创建或打开共享内存');
        }
    }

    /**
     * 写入数据
     */
    public function write(string $data): void
    {
        $size = strlen($data);
        if ($size > $this->size) {
            throw new RuntimeException("数据大小 ({$size}) 超过共享内存容量 ({$this->size})");
        }

        // 填充到固定长度
        $padded = str_pad($data, $this->size, "\0");

        shmop_write($this->shmId, $padded, 0);
    }

    /**
     * 读取数据
     */
    public function read(): string
    {
        $data = shmop_read($this->shmId, 0, $this->size);
        if ($data === false) {
            throw new RuntimeException('读取共享内存失败');
        }

        // 去除 null 填充
        return rtrim($data, "\0");
    }

    /**
     * 关闭共享内存
     */
    public function close(): void
    {
        if ($this->shmId !== 0) {
            shmop_close($this->shmId);
        }
    }

    /**
     * 删除共享内存
     */
    public function delete(): void
    {
        shmop_delete($this->shmId);
    }

    public function __destruct()
    {
        $this->close();
    }
}

// 进程1：写入
$shm = new ShmopManager('demo', 1024);
$shm->write('Hello from Process 1! Time: ' . date('H:i:s'));
echo "已写入共享内存" . PHP_EOL;

// 进程2：读取
// $shm = new ShmopManager('demo', 1024);
// echo "读取: " . $shm->read() . PHP_EOL;

使用共享内存的计数器 ​

php
<?php
declare(strict_types=1);

/**
 * 基于共享内存的原子计数器
 */
class SharedMemoryCounter
{
    private int $shmId;
    private int $semId;
    private const SIZE = 64;

    public function __construct(string $key = 'counter')
    {
        $shmKey = ftok(__FILE__, ord($key[0]));
        $semKey = $shmKey + 1;

        // 创建信号量用于同步
        $this->semId = sem_get($semKey, 1, 0666);
        if ($this->semId === false) {
            throw new RuntimeException('无法创建信号量');
        }

        // 创建共享内存
        $this->shmId = shmop_open($shmKey, 'c', 0644, self::SIZE);
        if ($this->shmId === false) {
            $this->shmId = shmop_open($shmKey, 'w', 0, 0);
        }

        if ($this->shmId === false) {
            throw new RuntimeException('无法创建共享内存');
        }

        // 初始化计数器
        $current = $this->readRaw();
        if ($current === '' || !is_numeric($current)) {
            $this->writeRaw('0');
        }
    }

    /**
     * 原子递增
     */
    public function increment(int $step = 1): int
    {
        // 获取信号量锁
        sem_acquire($this->semId);

        try {
            $count = (int) $this->readRaw() + $step;
            $this->writeRaw((string) $count);
            return $count;
        } finally {
            // 释放信号量锁
            sem_release($this->semId);
        }
    }

    /**
     * 获取当前值
     */
    public function get(): int
    {
        sem_acquire($this->semId);
        try {
            return (int) $this->readRaw();
        } finally {
            sem_release($this->semId);
        }
    }

    /**
     * 重置计数器
     */
    public function reset(): void
    {
        sem_acquire($this->semId);
        try {
            $this->writeRaw('0');
        } finally {
            sem_release($this->semId);
        }
    }

    private function readRaw(): string
    {
        $data = shmop_read($this->shmId, 0, self::SIZE);
        return rtrim($data !== false ? $data : '', "\0");
    }

    private function writeRaw(string $data): void
    {
        $padded = str_pad($data, self::SIZE, "\0");
        shmop_write($this->shmId, $padded, 0);
    }

    public function cleanup(): void
    {
        shmop_delete($this->shmId);
        sem_remove($this->semId);
    }
}

// 使用示例
$counter = new SharedMemoryCounter();

// 多次递增
for ($i = 0; $i < 100; $i++) {
    $counter->increment();
}

echo "当前计数: " . $counter->get() . PHP_EOL;

SysvSemaphore 信号量 ​

php
<?php
declare(strict_types=1);

/**
 * 信号量封装
 * 用于控制对共享资源的并发访问
 */
class SemaphoreGuard
{
    private int $semId;

    /**
     * 创建信号量
     */
    public function __construct(int $key, int $maxAcquire = 1)
    {
        $this->semId = sem_get($key, $maxAcquire, 0666);
        if ($this->semId === false) {
            throw new RuntimeException("无法创建信号量 (key={$key})");
        }
    }

    /**
     * 获取锁（阻塞等待）
     */
    public function acquire(): bool
    {
        return sem_acquire($this->semId);
    }

    /**
     * 尝试获取锁（非阻塞）
     */
    public function tryAcquire(): bool
    {
        return sem_acquire($this->semId, false);
    }

    /**
     * 释放锁
     */
    public function release(): bool
    {
        return sem_release($this->semId);
    }

    /**
     * 删除信号量
     */
    public function remove(): bool
    {
        return sem_remove($this->semId);
    }
}

/**
 * 使用 RAII 模式的锁
 */
class Lock implements \Stringable
{
    private SemaphoreGuard $semaphore;
    private bool $locked = false;

    public function __construct(int $key)
    {
        $this->semaphore = new SemaphoreGuard($key);
    }

    public function lock(): void
    {
        $this->semaphore->acquire();
        $this->locked = true;
    }

    public function unlock(): void
    {
        if ($this->locked) {
            $this->semaphore->release();
            $this->locked = false;
        }
    }

    public function __destruct()
    {
        $this->unlock();
    }

    public function __toString(): string
    {
        return $this->locked ? 'Locked' : 'Unlocked';
    }
}

// 使用示例
$lock = new Lock(12345);

try {
    echo "获取锁..." . PHP_EOL;
    $lock->lock();
    echo "已获取锁，执行临界区操作..." . PHP_EOL;

    sleep(2);

    echo "操作完成" . PHP_EOL;
} finally {
    $lock->unlock();
    // 或依赖析构函数自动解锁
}

PHP 8.0+ SharedMemory 扩展 ​

php
<?php
declare(strict_types=1);

/**
 * PHP 8.0+ SharedMemory 扩展使用
 * 提供了面向对象的共享内存 API
 */

// 创建共享内存对象（PHP 8.0+）
// 需要安装 ext-shmop 或 ext-sysvshm

/**
 * 使用 SysV 共享内存（面向对象接口）
 */
class SysvSharedMemoryExample
{
    public function create(): void
    {
        // 创建共享内存段
        $shmId = shm_attach(ftok(__FILE__, 't'), 1024, 0666);

        if ($shmId === false) {
            throw new RuntimeException('无法创建共享内存');
        }

        // 写入变量
        shm_put_var($shmId, 1, [
            'counter' => 0,
            'last_update' => time(),
            'data' => ['hello', 'world'],
        ]);

        // 读取变量
        $data = shm_get_var($shmId, 1);
        print_r($data);

        // 检查变量是否存在
        if (shm_has_var($shmId, 1)) {
            echo "变量存在" . PHP_EOL;
        }

        // 删除变量
        shm_remove_var($shmId, 1);

        // 关闭共享内存
        shm_detach($shmId);
    }

    /**
     * 共享内存 + 信号量实现安全的任务队列状态
     */
    public function taskQueueState(): void
    {
        $shmKey = ftok(__FILE__, 'q');
        $semKey = ftok(__FILE__, 's');

        $shm = shm_attach($shmKey, 4096, 0666);
        $sem = sem_get($semKey);

        // 加锁
        sem_acquire($sem);

        try {
            // 读取当前状态
            if (!shm_has_var($shm, 1)) {
                shm_put_var($shm, 1, [
                    'pending' => 0,
                    'processing' => 0,
                    'completed' => 0,
                    'failed' => 0,
                ]);
            }

            $state = shm_get_var($shm, 1);
            $state['pending']++;

            // 更新状态
            shm_put_var($shm, 1, $state);
        } finally {
            sem_release($sem);
        }

        shm_detach($shm);
    }
}

实战示例 ​

跨进程限流器 ​

php
<?php
declare(strict_types=1);

/**
 * 基于共享内存的速率限制器
 * 支持多进程间共享限流计数
 */
class SharedMemoryRateLimiter
{
    private int $shmId;
    private int $semId;
    private const SIZE = 128;
    private int $maxRequests;
    private int $windowSeconds;

    public function __construct(
        string $key,
        int $maxRequests = 100,
        int $windowSeconds = 60
    ) {
        $this->maxRequests = $maxRequests;
        $this->windowSeconds = $windowSeconds;

        $shmKey = ftok(__FILE__, ord($key[0]));
        $semKey = $shmKey + 1;

        $this->semId = sem_get($semKey, 1, 0666);
        $this->shmId = shmop_open($shmKey, 'c', 0644, self::SIZE);
    }

    /**
     * 检查是否允许请求
     */
    public function allow(): bool
    {
        sem_acquire($this->semId);

        try {
            $data = $this->readState();

            $now = time();
            $windowStart = $now - $this->windowSeconds;

            // 清理过期记录
            $data['timestamps'] = array_filter(
                $data['timestamps'],
                fn(int $ts) => $ts > $windowStart
            );

            if (count($data['timestamps']) >= $this->maxRequests) {
                return false;
            }

            $data['timestamps'][] = $now;
            $this->writeState($data);
            return true;
        } finally {
            sem_release($this->semId);
        }
    }

    /**
     * 获取剩余配额
     */
    public function getRemaining(): int
    {
        sem_acquire($this->semId);
        try {
            $data = $this->readState();
            $windowStart = time() - $this->windowSeconds;
            $data['timestamps'] = array_filter(
                $data['timestamps'],
                fn(int $ts) => $ts > $windowStart
            );
            return max(0, $this->maxRequests - count($data['timestamps']));
        } finally {
            sem_release($this->semId);
        }
    }

    private function readState(): array
    {
        $raw = shmop_read($this->shmId, 0, self::SIZE);
        $data = rtrim($raw !== false ? $raw : '', "\0");

        if ($data === '' || $data === "\0") {
            return ['timestamps' => []];
        }

        $decoded = unserialize($data);
        return is_array($decoded) ? $decoded : ['timestamps' => []];
    }

    private function writeState(array $data): void
    {
        $serialized = serialize($data);
        $padded = str_pad($serialized, self::SIZE, "\0");
        shmop_write($this->shmId, $padded, 0);
    }
}

// 使用示例
$limiter = new SharedMemoryRateLimiter('api', 10, 60);

for ($i = 0; $i < 15; $i++) {
    $allowed = $limiter->allow();
    echo "请求 #{$i}: " . ($allowed ? '允许' : '限流') . PHP_EOL;
}
echo "剩余配额: " . $limiter->getRemaining() . PHP_EOL;

多进程共享配置 ​

php
<?php
declare(strict_types=1);

/**
 * 使用共享内存缓存配置（避免重复读取文件）
 */
class SharedConfigCache
{
    private int $shmId;
    private int $semId;
    private const SIZE = 4096;
    private string $configPath;

    public function __construct(string $configPath, string $key = 'config')
    {
        $this->configPath = $configPath;

        $shmKey = ftok(__FILE__, ord($key[0]));
        $semKey = $shmKey + 1;

        $this->semId = sem_get($semKey, 1, 0666);
        $this->shmId = shmop_open($shmKey, 'c', 0644, self::SIZE);
    }

    /**
     * 获取配置（带文件修改检查）
     */
    public function getConfig(): array
    {
        sem_acquire($this->semId);

        try {
            $data = $this->readRaw();

            if ($data !== '' && $data !== "\0") {
                $cached = unserialize(rtrim($data, "\0"));
                $fileMtime = filemtime($this->configPath);

                if (is_array($cached) && isset($cached['_mtime'])
                    && $cached['_mtime'] === $fileMtime) {
                    unset($cached['_mtime']);
                    return $cached;
                }
            }

            // 重新加载配置
            $config = require $this->configPath;
            $config['_mtime'] = filemtime($this->configPath);
            $this->writeRaw(serialize($config));

            unset($config['_mtime']);
            return $config;
        } finally {
            sem_release($this->semId);
        }
    }

    private function readRaw(): string
    {
        $data = shmop_read($this->shmId, 0, self::SIZE);
        return $data !== false ? $data : '';
    }

    private function writeRaw(string $data): void
    {
        $padded = str_pad($data, self::SIZE, "\0");
        shmop_write($this->shmId, $padded, 0);
    }
}

注意事项 ​

内存管理 ​

  1. 手动清理：共享内存不会自动回收，进程退出后仍然存在
  2. 内存泄漏：忘记删除共享内存会导致系统内存逐渐耗尽
  3. 大小固定：创建时指定大小，之后不能改变
bash
# 查看系统中的共享内存段
ipcs -m

# 删除共享内存段
ipcrm -m <shmid>

# 查看系统中的信号量
ipcs -s

# 删除信号量
ipcrm -s <semid>

共享内存泄漏

生产环境中必须确保在进程异常退出时也能正确清理共享内存。建议使用 register_shutdown_function() 和信号处理器配合清理。

安全考虑 ​

  1. 权限控制：设置合适的文件权限（0666 vs 0660）
  2. 数据验证：从共享内存读取的数据必须进行验证
  3. 大小检查：写入前检查数据大小是否超出限制

最佳实践 ​

1. 使用封装类 ​

始终通过封装类操作共享内存，避免直接使用 shmop_* 函数。

2. 信号量保护所有共享内存访问 ​

所有读写共享内存的操作都必须在信号量保护下进行。

3. 考虑替代方案 ​

在现代 PHP 应用中，Redis 通常比共享内存更适合跨进程通信：

  • Redis 提供丰富的数据结构
  • Redis 支持过期机制
  • Redis 可以跨机器通信
  • Redis 有成熟的客户端库

仅在需要极低延迟的场景（微秒级）时才考虑直接使用共享内存。

下一节

继续学习：Swoole 协程框架

参考链接 ​