PHP 怎么订阅管理

本文目录导读：

1. 事件订阅（Event Subscription）
2. Redis 发布/订阅（消息队列场景）
3. 数据库驱动的订阅管理（用户订阅功能）
4. WebSocket 订阅（实时推送）
5. 使用成熟库（推荐）
6. 最佳实践建议
7. 选择建议

在 PHP 中，“订阅管理”通常指以下几种场景,我分别给出解决方案：

事件订阅（Event Subscription）

使用内置的观察者模式

<?php
// 定义事件接口
interface Subject {
    public function attach(Observer $observer);
    public function detach(Observer $observer);
    public function notify();
}
interface Observer {
    public function update(Subject $subject);
}
// 具体实现
class Order implements Subject {
    private $observers = [];
    private $status;
    public function attach(Observer $observer) {
        $this->observers[] = $observer;
    }
    public function detach(Observer $observer) {
        foreach ($this->observers as $key => $obs) {
            if ($obs === $observer) {
                unset($this->observers[$key]);
            }
        }
    }
    public function notify() {
        foreach ($this->observers as $observer) {
            $observer->update($this);
        }
    }
    public function setStatus($status) {
        $this->status = $status;
        $this->notify();
    }
    public function getStatus() {
        return $this->status;
    }
}
class EmailNotifier implements Observer {
    public function update(Subject $subject) {
        echo "发送邮件通知：订单状态变为 {$subject->getStatus()}\n";
    }
}
class SMSNotifier implements Observer {
    public function update(Subject $subject) {
        echo "发送短信通知：订单状态变为 {$subject->getStatus()}\n";
    }
}
// 使用
$order = new Order();
$order->attach(new EmailNotifier());
$order->attach(new SMSNotifier());
$order->setStatus('已发货');

Redis 发布/订阅（消息队列场景）

<?php
// 订阅端
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
// 订阅频道（阻塞方式）
$redis->subscribe(['channel1', 'channel2'], function($redis, $channel, $message) {
    echo "收到来自 {$channel} 的消息: {$message}\n";
});
// 或者使用 psubscribe 支持模式匹配
$redis->psubscribe(['news.*'], function($redis, $pattern, $channel, $message) {
    echo "模式 {$pattern} 匹配到 {$channel}: {$message}\n";
});
// 发布端
$redis->publish('channel1', 'Hello World');
$redis->publish('news.sports', '体育新闻');

数据库驱动的订阅管理（用户订阅功能）

<?php
class SubscriptionManager {
    private $pdo;
    public function __construct(PDO $pdo) {
        $this->pdo = $pdo;
    }
    // 创建订阅
    public function subscribe($userId, $planId, $duration) {
        $sql = "INSERT INTO subscriptions 
                (user_id, plan_id, start_date, end_date, status) 
                VALUES (?, ?, NOW(), DATE_ADD(NOW(), INTERVAL ? DAY), 'active')";
        $stmt = $this->pdo->prepare($sql);
        return $stmt->execute([$userId, $planId, $duration]);
    }
    // 取消订阅
    public function unsubscribe($userId) {
        $sql = "UPDATE subscriptions 
                SET status = 'cancelled', cancelled_at = NOW() 
                WHERE user_id = ? AND status = 'active'";
        $stmt = $this->pdo->prepare($sql);
        return $stmt->execute([$userId]);
    }
    // 检查订阅状态
    public function checkSubscription($userId) {
        $sql = "SELECT * FROM subscriptions 
                WHERE user_id = ? AND status = 'active' 
                AND end_date > NOW() 
                ORDER BY end_date DESC LIMIT 1";
        $stmt = $this->pdo->prepare($sql);
        $stmt->execute([$userId]);
        return $stmt->fetch(PDO::FETCH_ASSOC);
    }
    // 续订
    public function renew($userId, $days) {
        $current = $this->checkSubscription($userId);
        if ($current) {
            $sql = "UPDATE subscriptions SET end_date = DATE_ADD(end_date, INTERVAL ? DAY) WHERE id = ?";
            return $this->pdo->prepare($sql)->execute([$days, $current['id']]);
        }
        return $this->subscribe($userId, 1, $days);
    }
}

WebSocket 订阅（实时推送）

<?php
// 使用 Ratchet 库
use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;
class WebSocketSubscriber implements MessageComponentInterface {
    protected $clients;
    protected $subscriptions;
    public function __construct() {
        $this->clients = new \SplObjectStorage;
        $this->subscriptions = [];
    }
    public function onOpen(ConnectionInterface $conn) {
        $this->clients->attach($conn);
        echo "新连接: {$conn->resourceId}\n";
    }
    public function onMessage(ConnectionInterface $from, $msg) {
        $data = json_decode($msg, true);
        if (isset($data['action'])) {
            switch ($data['action']) {
                case 'subscribe':
                    $this->addSubscription($from, $data['channel']);
                    break;
                case 'unsubscribe':
                    $this->removeSubscription($from, $data['channel']);
                    break;
            }
        }
    }
    private function addSubscription($conn, $channel) {
        if (!isset($this->subscriptions[$channel])) {
            $this->subscriptions[$channel] = [];
        }
        $this->subscriptions[$channel][$conn->resourceId] = $conn;
    }
    private function removeSubscription($conn, $channel) {
        if (isset($this->subscriptions[$channel])) {
            unset($this->subscriptions[$channel][$conn->resourceId]);
        }
    }
    public function publish($channel, $message) {
        if (!isset($this->subscriptions[$channel])) return;
        foreach ($this->subscriptions[$channel] as $client) {
            $client->send(json_encode([
                'channel' => $channel,
                'message' => $message
            ]));
        }
    }
    public function onClose(ConnectionInterface $conn) {
        $this->clients->detach($conn);
        $this->cleanupSubscriptions($conn);
    }
    public function onError(ConnectionInterface $conn, \Exception $e) {
        echo "错误: {$e->getMessage()}\n";
        $conn->close();
    }
}

使用成熟库（推荐）

使用 RabbitMQ（php-amqplib）

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
// 发布订阅模式
function createPublisher() {
    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    // 声明交换器（fanout 广播模式）
    $channel->exchange_declare('notifications', 'fanout', false, false, false);
    $message = new AMQPMessage('Hello World!');
    $channel->basic_publish($message, 'notifications');
    $channel->close();
    $connection->close();
}
function createSubscriber() {
    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    $channel->exchange_declare('notifications', 'fanout', false, false, false);
    // 声明临时队列
    list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);
    // 绑定队列到交换器
    $channel->queue_bind($queue_name, 'notifications');
    echo "等待消息，退出请按 CTRL+C\n";
    $callback = function ($msg) {
        echo '收到消息: ', $msg->body, "\n";
    };
    $channel->basic_consume($queue_name, '', false, true, false, false, $callback);
    while ($channel->is_consuming()) {
        $channel->wait();
    }
}

最佳实践建议

订阅管理器类（综合方案）

<?php
class SubscriptionHandler {
    private $subscribers = [];
    private $persistence;
    public function __construct(PersistenceInterface $persistence) {
        $this->persistence = $persistence;
    }
    public function registerSubscriber($eventType, callable $callback, $id = null) {
        $subscriberId = $id ?? uniqid('sub_', true);
        $this->subscribers[$eventType][$subscriberId] = $callback;
        // 持久化存储
        $this->persistence->save('subscription', [
            'id' => $subscriberId,
            'event_type' => $eventType,
            'callback' => $callback
        ]);
        return $subscriberId;
    }
    public function unregisterSubscriber($eventType, $subscriberId) {
        if (isset($this->subscribers[$eventType][$subscriberId])) {
            unset($this->subscribers[$eventType][$subscriberId]);
            $this->persistence->delete('subscription', $subscriberId);
            return true;
        }
        return false;
    }
    public function triggerEvent($eventType, $payload = null) {
        if (!isset($this->subscribers[$eventType])) return;
        foreach ($this->subscribers[$eventType] as $callback) {
            call_user_func($callback, $payload);
        }
    }
    public function getSubscribers($eventType = null) {
        if ($eventType) {
            return $this->subscribers[$eventType] ?? [];
        }
        return $this->subscribers;
    }
}

选择建议

根据你的具体需求（是事件订阅、消息订阅、还是用户订阅管理）,选择最合适的方式实现即可。