PHP 怎么事件总线

wen PHP项目 2026-08-02 211

本文目录导读：

PHP 怎么事件总线

手动实现（轻量级事件总线）
改进版（支持异步队列、事件继承）
使用成熟库
最佳实践建议

在 PHP 中实现事件总线，通常有两种思路：手动实现（轻量级）和使用成熟库（功能丰富）,这里提供完整的实现方案。

手动实现（轻量级事件总线）

这是一个最基础的实现,适合中小型项目：

<?php
/** 事件接口 - 所有事件都实现此接口 */
interface Event { }
/** 监听器接口 */
interface Listener { 
    public function handle(Event $event): void; 
}
/** 事件总线核心类 */
final class EventBus {
    private static ?EventBus $instance = null;
    private array $listeners = [];
    // 单例模式
    public static function getInstance(): EventBus {
        if (self::$instance === null) {
            self::$instance = new self();
        }
        return self::$instance;
    }
    // 禁止外部实例化
    private function __construct() { }
    /** 注册监听器 */
    public function addListener(string $eventClass, Listener $listener, int $priority = 0): void {
        $this->listeners[$eventClass][$priority][] = $listener;
        // 按优先级排序
        ksort($this->listeners[$eventClass]);
    }
    /** 注册可调用监听器 */
    public function on(string $eventClass, callable $callback, int $priority = 0): void {
        $this->addListener($eventClass, new CallbackListener($callback), $priority);
    }
    /** 触发事件 */
    public function dispatch(Event $event): void {
        $eventClass = get_class($event);
        if (!isset($this->listeners[$eventClass])) {
            return;
        }
        foreach ($this->listeners[$eventClass] as $priority => $listeners) {
            foreach ($listeners as $listener) {
                $listener->handle($event);
            }
        }
    }
    /** 移除监听器 */
    public function removeListener(string $eventClass, Listener $listener): void {
        foreach ($this->listeners[$eventClass] ?? [] as $priority => $listeners) {
            $this->listeners[$eventClass][$priority] = array_filter(
                $listeners, 
                fn($l) => $l !== $listener
            );
        }
    }
}
/** 可调用监听器适配器 */
class CallbackListener implements Listener {
    private $callback;
    public function __construct(callable $callback) {
        $this->callback = $callback;
    }
    public function handle(Event $event): void {
        call_user_func($this->callback, $event);
    }
}
// ===== 使用示例 =====
/** 定义事件 */
class UserRegistered implements Event {
    public function __construct(public string $email, public string $name) { }
}
class OrderCreated implements Event {
    public function __construct(public int $orderId, public float $amount) { }
}
/** 定义监听器 */
class SendWelcomeEmail implements Listener {
    public function handle(Event $event): void {
        // 这里触发时不阻塞业务
        echo "📧 发送欢迎邮件到: {$event->email}\n";
    }
}
class CreateUserBin implements Listener {
    public function handle(Event $event): void {
        // 这里触发时不阻塞业务
        echo "📁 为用户创建初始目录\n";
    }
}
class SendOrderNotification implements Listener {
    public function handle(Event $event): void {
        echo "🔔 订单 #{$event->orderId} 金额会计入报表\n";
    }
}
// 业务代码
$bus = EventBus::getInstance();
// 注册监听器
$bus->addListener(UserRegistered::class, new SendWelcomeEmail(), 10);
$bus->addListener(UserRegistered::class, new CreateUserBin(), 20);
// 使用闭包
$bus->on(OrderCreated::class, function(OrderCreated $event) {
    echo "📦 创建订单通知: #{$event->orderId}\n";
});
// 业务代码中触发事件（不阻塞业务）
$bus->dispatch(new UserRegistered('huang@example.com', 'Huang'));
$bus->dispatch(new OrderCreated(1001, 299.99));

改进版（支持异步队列、事件继承）

<?php
interface Event { 
    public function getName(): string; // 事件名称
}
interface Listener {
    public function handle(Event $event): void;
}
/** 异步事件处理器 */
interface AsyncEventBus {
    /** 添加异步监听器，通过队列(如 Redis)处理 */
    public function addAsyncListener(string $eventName, string $queueName, Listener $listener): void;
    /** 触发异步事件 */
    public function dispatchAsync(Event $event, string $queueName): void;
}
final class AdvancedEventBus implements AsyncEventBus {
    private array $listeners = [];
    private array $asyncListeners = [];
    private array $emittedEvents = []; // 用于调试
    public function __construct(
        private ?EventQueueInterface $queue = null  // 队列实现（如 Redis）
    ) { }
    /** 同步监听 */
    public function addListener(string $eventName, Listener $listener, int $priority = 0): void {
        $this->listeners[$eventName][$priority][] = $listener;
    }
    /** 同步触发 */
    public function dispatch(Event $event): void {
        $eventName = $event->getName();
        $this->emittedEvents[] = $eventName . ' @ ' . date('H:i:s');
        // 处理父类事件
        foreach (class_parents($event) as $parentClass) {
            $this->handleListeners($parentClass, $event);
        }
        // 处理自身事件
        $this->handleListeners($eventName, $event);
    }
    private function handleListeners(string $eventName, Event $event): void {
        if (!isset($this->listeners[$eventName])) return;
        foreach ($this->listeners[$eventName] as $priority => $listeners) {
            foreach ($listeners as $listener) {
                try {
                    $listener->handle($event);
                } catch (\Throwable $e) {
                    // 记录错误，不让错误中断主流程
                    error_log("事件监听器错误: " . $e->getMessage());
                }
            }
        }
    }
    /** 异步触发 */
    public function dispatchAsync(Event $event, string $queueName = 'default'): void {
        if ($this->queue) {
            $this->queue->push($event, $queueName);
        }
    }
    /** 获取已触发事件（调试用） */
    public function getEmittedEvents(): array {
        return $this->emittedEvents;
    }
}

使用成熟库

Symfony EventDispatcher（推荐）

composer require symfony/event-dispatcher

use Symfony\Component\EventDispatcher\EventDispatcher;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Contracts\EventDispatcher\Event;
// 定义事件
class OrderPlaced extends Event {
    public const NAME = 'order.placed';
    public function __construct(
        public readonly int $orderId,
        public readonly string $orderNumber
    ) { }
}
// 监听器
class SendOrderEmail {
    public function __invoke(OrderPlaced $event): void {
        echo "📨 发送订单邮件: {$event->orderNumber}\n";
    }
}
// 注册和使用
$dispatcher = new EventDispatcher();
$dispatcher->addListener(OrderPlaced::NAME, new SendOrderEmail());
// 或使用订阅者
class OrderSubscriber implements EventSubscriberInterface {
    public static function getSubscribedEvents(): array {
        return [
            OrderPlaced::NAME => [['sendEmail', 10], ['updateInventory', 20]],
        ];
    }
    public function sendEmail(OrderPlaced $event): void { /* ... */ }
    public function updateInventory(OrderPlaced $event): void { /* ... */ }
}
// 业务代码中触发
$dispatcher->dispatch(new OrderPlaced(1001, 'ORDER-2024-1001'), OrderPlaced::NAME);

League/Event（轻量级）

composer require league/event

use League\Event\Emitter;
use League\Event\Listener;
$emitter = new Emitter();
// 监听
$emitter->addListener('user.registered', new class implements Listener {
    public function handle($event): void {
        echo "用户注册了: {$event->email}\n";
    }
});
// 触发
$emitter->emit('user.registered', (object)['email' => 'user@example.com']);

最佳实践建议

生产环境方案（完整示例）

<?php
// event-bus.php
use Symfony\Component\EventDispatcher\EventDispatcher;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
// 1. 定义事件基类
abstract class BaseEvent {
    public const NAME = '';
}
// 2. 具体事件
class UserRegisteredEvent extends BaseEvent {
    public const NAME = 'user.registered';
    public function __construct(
        public readonly string $userId,
        public readonly string $email
    ) { }
}
// 3. 监听器接口
interface EventListener {
    public function handle(BaseEvent $event): void;
}
// 4. 事件总线封装（单例）
final class AppEventBus {
    private static ?AppEventBus $instance = null;
    private EventDispatcher $dispatcher;
    private array $queuedEvents = []; // 延迟处理的事件
    private function __construct() {
        $this->dispatcher = new EventDispatcher();
        $this->registerDefaultListeners();
    }
    public static function getInstance(): AppEventBus {
        if (self::$instance === null) {
            self::$instance = new AppEventBus();
        }
        return self::$instance;
    }
    /** 注册默认监听器 */
    private function registerDefaultListeners(): void {
        // 注册日志监听
        $this->dispatcher->addListener(UserRegisteredEvent::NAME, function($event) {
            $this->log("事件触发: " . $event::class . " 用户: " . $event->email);
        });
    }
    /** 触发事件 */
    public function emit(BaseEvent $event): void {
        $this->dispatcher->dispatch($event, $event::NAME);
    }
    /** 延迟到请求结束后触发 */
    public function emitLater(BaseEvent $event): void {
        $this->queuedEvents[] = $event;
    }
    /** 处理延迟事件（通常在框架的 shutdown 钩子中调用） */
    public function flushQueuedEvents(): void {
        foreach ($this->queuedEvents as $event) {
            $this->emit($event);
        }
        $this->queuedEvents = [];
    }
    private function log(string $message): void {
        // 记录到日志文件
        file_put_contents(
            __DIR__ . '/events.log', 
            date('Y-m-d H:i:s') . " - " . $message . "\n", 
            FILE_APPEND
        );
    }
}
// 5. 业务代码中的使用
$bus = AppEventBus::getInstance();
// 用户注册流程
$registerUser = function(string $email) use ($bus) {
    // 业务逻辑：保存用户...
    $userId = 123;
    // 触发事件（不阻塞业务）
    $bus->emit(new UserRegisteredEvent($userId, $email));
};
$registerUser('newuser@example.com');
$bus->flushQueuedEvents(); // 处理延迟事件（在框架生命周期末尾调用）

性能优化建议

避免同步阻塞：对于邮件、短信等耗时操作，使用异步队列
优先级控制：关键逻辑监听器优先执行
异常隔离：监听器内部异常不影响主逻辑
事件命名规范：对象.动作 格式，如 user.registered、order.paid

选择建议

场景	推荐方案
小型项目、快速开发	手动实现的轻量事件总线
中型项目、有明确事件分类	Symfony EventDispatcher
大型项目、需要集群广播	使用 Redis Pub/Sub 配合
需要异步处理	加上队列（RabbitMQ、Kafka）

选择适合自己项目的方案，推荐使用 Symfony EventDispatcher（成熟稳定、功能完善、文档丰富），如果是小型项目,手动实现一个轻量级的也完全够用。

本文地址： https://www.bmjup.com/post/36354.html