本文目录导读：
在 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;
}
}
选择建议
| 使用场景 | 推荐方案 |
|---|---|
| 简单事件通知 | 内置观察者模式 |
| 跨进程/跨服务通信 | Redis Pub/Sub 或 RabbitMQ |
| 网站订阅功能 | 数据库实现 + 定时任务 |
| 实时推送 | WebSocket + 消息队列 |
| 复杂事件驱动架构 | Symfony EventDispatcher 或 Laravel Events |
根据你的具体需求（是事件订阅、消息订阅、还是用户订阅管理）,选择最合适的方式实现即可。
