汤不热吧什么是事件驱动架构
事件驱动架构（Event-Driven Architecture, EDA）是一种以事件的产生、检测和消费为核心的软件架构范式。在传统 PHP 应用中，我们习惯于”请求-响应”模型：用户发起请求，控制器调用模型处理，返回视图。但当业务复杂度增长时，这种模型会导致控制器臃肿、模块耦合严重、扩展困难。
事件驱动架构的核心思想是：系统中的一切重要变化都表现为事件。一个事件是”已经发生的事实”，比如”用户已注册”、”订单已创建”、”支付已完成”。这些事件一旦产生就不可变，其他模块通过订阅和响应这些事件来完成各自的工作，而无需了解事件的生产者。
在 PHP 生态中，Laravel 的 Event 系统、Symfony 的 EventDispatcher 组件、以及 PSR-14 事件分发器标准都为事件驱动架构提供了基础设施。但真正用好事件驱动架构，需要理解更深层的概念：领域事件、事件溯源和 CQRS。
领域事件：业务语义的精确表达
领域事件（Domain Event）是领域驱动设计（DDD）中的核心概念。它不是技术层面的”按钮点击”或”HTTP 请求”，而是用业务语言描述的、已经发生的业务事实。
领域事件的定义
一个好的领域事件应该：
- 用过去时态命名（
1UserRegistered
而非
1RegisterUser）
- 携带足够的上下文信息，让订阅者无需回查
- 是不可变的——事件一旦发生就不能被修改或撤销
- 具有业务含义——反映真实发生的业务变更
下面是一个领域事件的基础实现：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
declare(strict_types=1);
namespace App\Domain\Event;
use DateTimeImmutable;
abstract class DomainEvent
{
public readonly DateTimeImmutable $occurredAt;
public readonly string $eventId;
public function __construct()
{
$this->occurredAt = new DateTimeImmutable();
$this->eventId = bin2hex(random_bytes(16));
}
abstract public function eventName(): string;
abstract public function toArray(): array;
public static function fromArray(array $data): static
{
return new static(...$data);
}
}
具体领域事件示例
以电商系统为例，定义几个核心领域事件：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
declare(strict_types=1);
namespace App\Domain\Event\Order;
use App\Domain\Event\DomainEvent;
class OrderCreated extends DomainEvent
{
public function __construct(
public readonly string $orderId,
public readonly string $customerId,
public readonly array $items,
public readonly float $totalAmount,
public readonly string $currency = 'CNY'
) {
parent::__construct();
}
public function eventName(): string
{
return 'order.created';
}
public function toArray(): array
{
return [
'orderId' => $this->orderId,
'customerId' => $this->customerId,
'items' => $this->items,
'totalAmount' => $this->totalAmount,
'currency' => $this->currency,
'occurredAt' => $this->occurredAt->format('c'),
'eventId' => $this->eventId,
];
}
}
class OrderPaid extends DomainEvent
{
public function __construct(
public readonly string $orderId,
public readonly string $paymentId,
public readonly float $paidAmount,
public readonly string $paymentMethod
) {
parent::__construct();
}
public function eventName(): string
{
return 'order.paid';
}
public function toArray(): array
{
return [
'orderId' => $this->orderId,
'paymentId' => $this->paymentId,
'paidAmount' => $this->paidAmount,
'paymentMethod' => $this->paymentMethod,
'occurredAt' => $this->occurredAt->format('c'),
'eventId' => $this->eventId,
];
}
}
class OrderShipped extends DomainEvent
{
public function __construct(
public readonly string $orderId,
public readonly string $trackingNumber,
public readonly string $carrier
) {
parent::__construct();
}
public function eventName(): string
{
return 'order.shipped';
}
public function toArray(): array
{
return [
'orderId' => $this->orderId,
'trackingNumber' => $this->trackingNumber,
'carrier' => $this->carrier,
'occurredAt' => $this->occurredAt->format('c'),
'eventId' => $this->eventId,
];
}
}
事件分发器：PSR-14 标准实现
PSR-14 是 PHP 标准推荐规范中关于事件分发器的标准。它定义了两个核心接口：
1 | EventDispatcherInterface |
和
1 | ListenerProviderInterface |
。这个标准让事件系统可以跨框架互换。
实现 PSR-14 事件分发器
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
declare(strict_types=1);
namespace App\EventDispatcher;
use Psr\EventDispatcher\EventDispatcherInterface;
use Psr\EventDispatcher\ListenerProviderInterface;
use Psr\EventDispatcher\StoppableEventInterface;
class EventDispatcher implements EventDispatcherInterface
{
public function __construct(
private readonly ListenerProviderInterface $listenerProvider
) {}
public function dispatch(object $event): object
{
$listeners = $this->listenerProvider->getListenersForEvent($event);
foreach ($listeners as $listener) {
if ($event instanceof StoppableEventInterface
&& $event->isPropagationStopped()) {
break;
}
$listener($event);
}
return $event;
}
}
class ListenerProvider implements ListenerProviderInterface
{
private array $listeners = [];
public function addListener(string $eventName, callable $listener): void
{
$this->listeners[$eventName][] = $listener;
}
public function getListenersForEvent(object $event): iterable
{
$className = get_class($event);
$interfaces = class_implements($event);
$listeners = [];
if (isset($this->listeners[$className])) {
$listeners = array_merge($listeners, $this->listeners[$className]);
}
if ($interfaces) {
foreach ($interfaces as $interface) {
if (isset($this->listeners[$interface])) {
$listeners = array_merge($listeners, $this->listeners[$interface]);
}
}
}
if (method_exists($event, 'eventName')) {
$eventName = $event->eventName();
if (isset($this->listeners[$eventName])) {
$listeners = array_merge($listeners, $this->listeners[$eventName]);
}
}
return $listeners;
}
}
注册事件监听器
在实际项目中，我们需要将业务逻辑组织为独立的事件处理器：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
declare(strict_types=1);
namespace App\EventListener;
use App\Domain\Event\Order\OrderCreated;
use App\Domain\Event\Order\OrderPaid;
class SendOrderConfirmationEmail
{
public function __construct(
private readonly MailerInterface $mailer,
private readonly LoggerInterface $logger
) {}
public function __invoke(OrderCreated $event): void
{
$this->logger->info('Sending order confirmation email', [
'orderId' => $event->orderId,
]);
$this->mailer->send(
to: $event->customerId,
subject: "订单确认 #{$event->orderId}",
body: $this->renderTemplate($event)
);
}
}
class UpdateInventory
{
public function __construct(
private readonly InventoryRepository $inventory
) {}
public function __invoke(OrderCreated $event): void
{
foreach ($event->items as $item) {
$this->inventory->decrement($item['sku'], $item['quantity']);
}
}
}
class SendPaymentReceipt
{
public function __invoke(OrderPaid $event): void
{
// 发送支付回执
}
}
// 在服务容器中注册
$provider->addListener('order.created', new SendOrderConfirmationEmail($mailer, $logger));
$provider->addListener('order.created', new UpdateInventory($inventory));
$provider->addListener('order.paid', new SendPaymentReceipt());
事件溯源：用事件日志替代状态存储
传统 CRUD 应用存储的是实体的当前状态——比如
1 | orders |
表中的
1 | status = 'paid' |
。但状态存储丢失了所有历史信息：我们不知道订单何时创建、何时支付、由谁修改。
事件溯源（Event Sourcing）的核心思想是：不存储实体的当前状态，而是存储所有改变状态的事件。实体的当前状态可以通过重放所有事件来重建。
事件存储实现
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
declare(strict_types=1);
namespace App\EventStore;
use App\Domain\Event\DomainEvent;
class EventStore
{
public function __construct(
private readonly PDO $pdo
) {}
public function append(string $aggregateId, DomainEvent $event, int $expectedVersion): void
{
$this->pdo->beginTransaction();
try {
// 乐观并发控制
$stmt = $this->pdo->prepare(
'SELECT MAX(version) as current_version FROM events WHERE aggregate_id = ?'
);
$stmt->execute([$aggregateId]);
$current = (int) $stmt->fetchColumn();
if ($current !== $expectedVersion) {
throw new ConcurrencyException(
"Expected version {$expectedVersion}, but got {$current}"
);
}
$newVersion = $expectedVersion + 1;
$stmt = $this->pdo->prepare("
INSERT INTO events (event_id, aggregate_id, event_type, payload, version, occurred_at)
VALUES (?, ?, ?, ?, ?, ?)
");
$stmt->execute([
$event->eventId,
$aggregateId,
$event->eventName(),
json_encode($event->toArray()),
$newVersion,
$event->occurredAt->format('Y-m-d H:i:s.u'),
]);
$this->pdo->commit();
} catch (\Throwable $e) {
$this->pdo->rollBack();
throw $e;
}
}
public function getEventsForAggregate(string $aggregateId): array
{
$stmt = $this->pdo->prepare("
SELECT event_type, payload FROM events
WHERE aggregate_id = ?
ORDER BY version ASC
");
$stmt->execute([$aggregateId]);
$events = [];
while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
$events[] = $this->hydrateEvent($row['event_type'], json_decode($row['payload'], true));
}
return $events;
}
public function getEventsAfter(int $lastEventId, int $limit = 100): array
{
$stmt = $this->pdo->prepare("
SELECT id, event_type, aggregate_id, payload FROM events
WHERE id > ?
ORDER BY id ASC
LIMIT ?
");
$stmt->execute([$lastEventId, $limit]);
$events = [];
while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
$events[] = $this->hydrateEvent($row['event_type'], json_decode($row['payload'], true));
}
return $events;
}
private function hydrateEvent(string $eventType, array $payload): DomainEvent
{
$classMap = [
'order.created' => \App\Domain\Event\Order\OrderCreated::class,
'order.paid' => \App\Domain\Event\Order\OrderPaid::class,
'order.shipped' => \App\Domain\Event\Order\OrderShipped::class,
];
if (!isset($classMap[$eventType])) {
throw new \RuntimeException("Unknown event type: {$eventType}");
}
return $classMap[$eventType]::fromArray($payload);
}
}
事件存储表结构
2
3
4
5
6
7
8
9
10
11
12
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
event_id VARCHAR(32) NOT NULL UNIQUE,
aggregate_id VARCHAR(36) NOT NULL,
event_type VARCHAR(100) NOT NULL,
payload JSON NOT NULL,
version INT UNSIGNED NOT NULL,
occurred_at DATETIME(6) NOT NULL,
INDEX idx_aggregate_id_version (aggregate_id, version),
INDEX idx_event_type (event_type),
INDEX idx_occurred_at (occurred_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
聚合根：从事件重建状态
聚合根（Aggregate Root）是 DDD 中的核心概念，它是一致性边界。在事件溯源架构中，聚合根不仅负责执行业务逻辑和验证，还负责通过重放事件来重建自身状态。
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
declare(strict_types=1);
namespace App\Domain\Order;
use App\Domain\Event\DomainEvent;
use App\Domain\Event\Order\OrderCreated;
use App\Domain\Event\Order\OrderPaid;
use App\Domain\Event\Order\OrderShipped;
class Order
{
private string $orderId;
private string $customerId;
private array $items = [];
private float $totalAmount = 0;
private string $status = 'draft';
private int $version = 0;
/** @var DomainEvent[] */
private array $pendingEvents = [];
public static function reconstitute(array $events): self
{
$order = new self();
foreach ($events as $event) {
$order->apply($event);
}
return $order;
}
public static function create(
string $orderId,
string $customerId,
array $items
): self {
$order = new self();
if (empty($items)) {
throw new \InvalidArgumentException('订单不能为空');
}
$total = array_reduce($items, fn(float $sum, array $item) =>
$sum + ($item['price'] * $item['quantity']), 0.0
);
$event = new OrderCreated($orderId, $customerId, $items, $total);
$order->recordAndApply($event);
return $order;
}
public function pay(string $paymentId, string $method): void
{
if ($this->status !== 'created') {
throw new \LogicException('只有已创建的订单才能支付');
}
$event = new OrderPaid($this->orderId, $paymentId, $this->totalAmount, $method);
$this->recordAndApply($event);
}
public function ship(string $trackingNumber, string $carrier): void
{
if ($this->status !== 'paid') {
throw new \LogicException('只有已支付的订单才能发货');
}
$event = new OrderShipped($this->orderId, $trackingNumber, $carrier);
$this->recordAndApply($event);
}
private function recordAndApply(DomainEvent $event): void
{
$this->pendingEvents[] = $event;
$this->apply($event);
}
private function apply(DomainEvent $event): void
{
match (get_class($event)) {
OrderCreated::class => $this->applyOrderCreated($event),
OrderPaid::class => $this->applyOrderPaid($event),
OrderShipped::class => $this->applyOrderShipped($event),
default => throw new \RuntimeException(
'Unknown event: ' . get_class($event)
),
};
}
private function applyOrderCreated(OrderCreated $event): void
{
$this->orderId = $event->orderId;
$this->customerId = $event->customerId;
$this->items = $event->items;
$this->totalAmount = $event->totalAmount;
$this->status = 'created';
$this->version++;
}
private function applyOrderPaid(OrderPaid $event): void
{
$this->status = 'paid';
$this->version++;
}
private function applyOrderShipped(OrderShipped $event): void
{
$this->status = 'shipped';
$this->version++;
}
public function releaseEvents(): array
{
$events = $this->pendingEvents;
$this->pendingEvents = [];
return $events;
}
public function getVersion(): int
{
return $this->version;
}
public function getStatus(): string
{
return $this->status;
}
}
CQRS：命令与查询职责分离
CQRS（Command Query Responsibility Segregation）是事件驱动架构的自然搭档。它的核心思想是将写操作（命令）和读操作（查询）分离到不同的模型中。
在传统架构中，同一个 Repository 既负责写入也负责读取。但在事件溯源架构中，聚合根只负责写入（通过事件），读取则需要通过”投影”（Projection）构建专门的读模型。
命令端（写模型）
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
declare(strict_types=1);
namespace App\Command;
use App\Domain\Order\Order;
use App\EventStore\EventStore;
class OrderCommandHandler
{
public function __construct(
private readonly EventStore $eventStore
) {}
public function handleCreateOrder(CreateOrderCommand $cmd): void
{
$order = Order::create(
$cmd->orderId,
$cmd->customerId,
$cmd->items
);
$this->saveAggregate($order);
}
public function handlePayOrder(PayOrderCommand $cmd): void
{
$events = $this->eventStore->getEventsForAggregate($cmd->orderId);
$order = Order::reconstitute($events);
$order->pay($cmd->paymentId, $cmd->paymentMethod);
$this->saveAggregate($order);
}
private function saveAggregate(Order $order): void
{
foreach ($order->releaseEvents() as $event) {
$this->eventStore->append(
$order->getOrderId(),
$event,
$order->getVersion() - 1
);
}
}
}
查询端（读模型）
读模型通过订阅事件来维护。每当有新事件产生，投影处理器就更新读模型表：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
declare(strict_types=1);
namespace App\Projection;
use App\Domain\Event\Order\OrderCreated;
use App\Domain\Event\Order\OrderPaid;
use App\Domain\Event\Order\OrderShipped;
class OrderProjection
{
public function __construct(
private readonly PDO $pdo
) {}
public function handleOrderCreated(OrderCreated $event): void
{
$stmt = $this->pdo->prepare("
INSERT INTO order_read_model (order_id, customer_id, status, total_amount, created_at)
VALUES (?, ?, 'created', ?, ?)
");
$stmt->execute([
$event->orderId,
$event->customerId,
$event->totalAmount,
$event->occurredAt->format('Y-m-d H:i:s'),
]);
}
public function handleOrderPaid(OrderPaid $event): void
{
$stmt = $this->pdo->prepare("
UPDATE order_read_model
SET status = 'paid', paid_at = ?, payment_method = ?
WHERE order_id = ?
");
$stmt->execute([
$event->occurredAt->format('Y-m-d H:i:s'),
$event->paymentMethod,
$event->orderId,
]);
}
public function handleOrderShipped(OrderShipped $event): void
{
$stmt = $this->pdo->prepare("
UPDATE order_read_model
SET status = 'shipped', tracking_number = ?, carrier = ?, shipped_at = ?
WHERE order_id = ?
");
$stmt->execute([
$event->trackingNumber,
$event->carrier,
$event->occurredAt->format('Y-m-d H:i:s'),
$event->orderId,
]);
}
}
读模型表设计为查询友好：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
order_id VARCHAR(36) PRIMARY KEY,
customer_id VARCHAR(36) NOT NULL,
status VARCHAR(20) NOT NULL,
total_amount DECIMAL(10,2) NOT NULL,
payment_method VARCHAR(30),
tracking_number VARCHAR(50),
carrier VARCHAR(30),
created_at DATETIME NOT NULL,
paid_at DATETIME,
shipped_at DATETIME,
INDEX idx_customer (customer_id),
INDEX idx_status (status),
INDEX idx_created_at (created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
快照：优化长事件流的重建性能
当一个聚合根有大量事件时，每次都从第一个事件开始重放会很慢。快照机制通过定期保存聚合根的序列化状态来解决这个问题：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
declare(strict_types=1);
namespace App\EventStore;
class SnapshotStore
{
private const SNAPSHOT_INTERVAL = 50;
public function __construct(
private readonly PDO $pdo
) {}
public function saveSnapshot(string $aggregateId, object $aggregate, int $version): void
{
$stmt = $this->pdo->prepare("
INSERT INTO snapshots (aggregate_id, version, state, created_at)
VALUES (?, ?, ?, NOW())
ON DUPLICATE KEY UPDATE version = ?, state = ?, created_at = NOW()
");
$state = serialize($aggregate);
$stmt->execute([$aggregateId, $version, $state, $version, $state]);
}
public function loadSnapshot(string $aggregateId): ?array
{
$stmt = $this->pdo->prepare("
SELECT version, state FROM snapshots
WHERE aggregate_id = ?
ORDER BY version DESC
LIMIT 1
");
$stmt->execute([$aggregateId]);
$row = $stmt->fetch(PDO::FETCH_ASSOC);
if (!$row) {
return null;
}
return [
'version' => (int) $row['version'],
'state' => unserialize($row['state']),
];
}
public function shouldCreateSnapshot(int $currentVersion): bool
{
return $currentVersion > 0 && $currentVersion % self::SNAPSHOT_INTERVAL === 0;
}
}
结合快照的聚合根重建流程：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
declare(strict_types=1);
namespace App\Domain\Order;
class OrderRepository
{
public function __construct(
private readonly EventStore $eventStore,
private readonly SnapshotStore $snapshotStore
) {}
public function load(string $orderId): Order
{
$snapshot = $this->snapshotStore->loadSnapshot($orderId);
if ($snapshot) {
$order = $snapshot['state'];
$fromVersion = $snapshot['version'] + 1;
} else {
$order = new Order();
$fromVersion = 0;
}
$events = $this->eventStore->getEventsFromVersion($orderId, $fromVersion);
foreach ($events as $event) {
$order->apply($event);
}
if ($this->snapshotStore->shouldCreateSnapshot($order->getVersion())) {
$this->snapshotStore->saveSnapshot($orderId, $order, $order->getVersion());
}
return $order;
}
}
事件总线与异步处理
在生产环境中，事件处理通常需要异步执行，以避免阻塞主业务流程。我们通过事件总线将领域事件分发到消息队列，实现异步处理和解耦：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
declare(strict_types=1);
namespace App\EventBus;
use App\Domain\Event\DomainEvent;
class EventBus
{
private array $handlers = [];
public function __construct(
private readonly Redis $redis,
private readonly EventDispatcher $dispatcher,
private readonly LoggerInterface $logger
) {}
public function publish(DomainEvent $event): void
{
// 1. 同步分发给本地监听器
$this->dispatcher->dispatch($event);
// 2. 异步入队，供其他服务消费
$this->redis->lpush(
'events:pending',
json_encode([
'eventId' => $event->eventId,
'eventType' => $event->eventName(),
'payload' => $event->toArray(),
'timestamp' => time(),
])
);
$this->logger->info('Event published', [
'eventId' => $event->eventId,
'eventType' => $event->eventName(),
]);
}
public function consume(int $batchSize = 10): void
{
while (true) {
$raw = $this->redis->rpop('events:pending');
if (!$raw) {
usleep(100000);
continue;
}
$data = json_decode($raw, true);
try {
$event = $this->hydrateEvent($data['eventType'], $data['payload']);
$this->processEvent($event);
} catch (\Throwable $e) {
$this->logger->error('Event processing failed', [
'eventId' => $data['eventId'],
'error' => $e->getMessage(),
]);
$this->redis->lpush('events:failed', $raw);
}
}
}
private function processEvent(DomainEvent $event): void
{
$handlers = $this->getHandlersForEvent($event->eventName());
foreach ($handlers as $handler) {
$handler($event);
}
}
}
实战案例：完整的订单处理流程
将上述所有组件组合起来，一个完整的订单处理流程如下：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
declare(strict_types=1);
namespace App\Controller;
class OrderController
{
public function __construct(
private readonly OrderCommandHandler $commandHandler,
private readonly OrderQueryService $queryService,
private readonly EventBus $eventBus
) {}
public function create(ServerRequestInterface $request): Response
{
$data = $request->getParsedBody();
$cmd = new CreateOrderCommand(
orderId: Uuid::uuid4()->toString(),
customerId: $data['customer_id'],
items: $data['items']
);
$this->commandHandler->handleCreateOrder($cmd);
return new JsonResponse([
'orderId' => $cmd->orderId,
'status' => 'created',
], 201);
}
public function show(string $orderId): Response
{
$order = $this->queryService->getOrder($orderId);
if (!$order) {
return new JsonResponse(['error' => 'Order not found'], 404);
}
return new JsonResponse($order);
}
public function list(ServerRequestInterface $request): Response
{
$params = $request->getQueryParams();
$orders = $this->queryService->listOrders(
status: $params['status'] ?? null,
page: (int) ($params['page'] ?? 1),
perPage: (int) ($params['per_page'] ?? 20)
);
return new JsonResponse([
'data' => $orders,
'total' => $this->queryService->countOrders($params['status'] ?? null),
]);
}
}
最佳实践与常见陷阱
事件粒度的选择
事件太细会导致事件数量爆炸和重放缓慢；事件太粗则丢失了有价值的业务细节。一个实用的原则是：事件的粒度应该与业务操作的原子性一致。”订单创建”是一个事件，而不是拆分成”订单头创建”+”订单行创建”等多个事件。
最终一致性
事件驱动架构天然是最终一致性的。写入事件后，读模型不会立即更新。这意味着用户下单后立刻查询，可能还看不到新订单。对于 PHP Web 应用，有几种应对策略：
- 同步更新投影——在请求内同步处理事件，保证读一致性
- 延迟刷新——前端在写操作后等待几百毫秒再查询
- 版本号检查——客户端带上写操作返回的版本号，读模型版本匹配后才返回
事件版本化
事件一旦存储就不可修改，但业务需求会变化。当事件结构需要升级时，使用”向上转换器”（Upcaster）：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
declare(strict_types=1);
namespace App\EventStore\Upcaster;
class OrderCreatedUpcaster
{
public function upcast(array $eventV1): array
{
// v1 没有 currency 字段，v2 添加了 currency
$eventV1['currency'] = $eventV1['currency'] ?? 'CNY';
// v1 用 amount，v2 改为 totalAmount
$eventV1['totalAmount'] = $eventV1['totalAmount'] ?? $eventV1['amount'] ?? 0;
unset($eventV1['amount']);
$eventV1['_version'] = 2;
return $eventV1;
}
}
幂等处理
事件可能被重复投递（比如消息队列的重试机制），处理器必须是幂等的。最简单的方式是利用事件 ID 去重：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
declare(strict_types=1);
class IdempotentEventHandler
{
public function __construct(
private readonly Redis $redis
) {}
public function handle(DomainEvent $event, callable $handler): void
{
$key = "processed:{$event->eventId}";
if ($this->redis->setnx($key, '1')) {
$this->redis->expire($key, 86400);
$handler($event);
}
}
}
何时使用事件溯源
事件溯源并不适合所有项目。它的引入增加了系统复杂度，需要投入更多的基础设施和运维成本。以下是适用和不适用场景的对比：
| 适用场景 | 不适用场景 |
|---|---|
| 需要完整审计追踪的金融/医疗系统 | 简单的 CRUD 应用 |
| 业务规则复杂、频繁变更的领域 | 读多写少、无复杂业务逻辑的展示型应用 |
| 需要时间旅行调试（回溯到任意时间点的状态） | 团队缺乏 DDD 经验 |
| 微服务间需要事件驱动的数据同步 | 对实时一致性要求极高的场景 |
| 需要构建多个不同视角的读模型 | 数据量极大且事件流极长的实体 |
总结
事件驱动架构与事件溯源为 PHP 应用带来了强大的解耦能力和完整的业务历史追溯能力。核心要点回顾：
- 领域事件是业务事实的精确表达，用过去时态命名，携带充分上下文
- PSR-14 提供了标准化的事件分发接口，可与任何框架集成
- 事件溯源用事件流替代状态存储，配合乐观并发控制保证一致性
- 聚合根通过重放事件重建状态，集中管理业务规则和不变量
- CQRS 分离读写模型，写端用事件溯源，读端用投影优化的查询表
- 快照解决长事件流的重放性能问题
- 事件总线实现异步处理和服务间解耦
- 幂等处理和版本化是生产环境不可或缺的保障
从实际出发，不必一步到位引入完整的事件溯源。可以先用 PSR-14 事件分发器解耦模块，再逐步引入领域事件和投影，最后在核心聚合根上实施事件溯源。渐进式演进比推倒重来更现实，也更安全。