Skip to content

Commit 1b9f1f9

Browse files
youxingxiangyouxingxiang
committed
laravel-event-sourcing 源码解析
1 parent ffa4726 commit 1b9f1f9

File tree

1 file changed

+369
-0
lines changed

1 file changed

+369
-0
lines changed

‎yxx_laravel_event_sourcing.md

Lines changed: 369 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,369 @@
1+
## 思考
2+
[Laravel-Event-Sourcing](https://spatie.be/docs/laravel-event-sourcing/v4/introduction) 中 StoredEvent 怎么保存 而 Projector 和 Reactor 又是怎么样监听到对应的事件的？
3+
## 案例
4+
用户取消订单,需要触发 `BookingCancelled` 事件，我们需要在 `Projector` 监听到 `BookingCancelled` 后改变订单状态。同时需要在 `Reactor` 发送一封邮件给用户。
5+
* 我们有一个 AggregateRoot `App\Bookings\AggregateRoot\BookingRoot`。里面有个取消订单的方法 cancle 代码如下：
6+
```php
7+
class BookingRoot extends AggregateRoot
8+
public function cancel(): self
9+
{
10+
$this->recordThat(
11+
new BookingCancelled(
12+
$this->userId,
13+
)
14+
);
15+
return $this;
16+
}
17+
}
18+
```
19+
20+
* 我们有一个 Projector `App\Bookings\Projector\BookingsProjector` 。他是同步的，代码如下：
21+
```php
22+
class BookingsProjector extends Projector
23+
{
24+
public function onBookingCancelled(
25+
StoredEvent $storedEvent,
26+
BookingCancelled $event,
27+
string $aggregateUuid
28+
) {
29+
$projection = ProjectionBooking::query()
30+
->where('booking_aggregate_id', $aggregateUuid)
31+
->first();
32+
$projection->status = BookingStatus::STATUS_CANCELLED;
33+
$projection->cancelled = $storedEvent->created_at;
34+
$projection->save();
35+
}
36+
}
37+
```
38+
* 我们有一个 Reactor `App\Bookings\Reactor\NotificationReactor`。他是异步的,代码如下：
39+
```php
40+
class NotificationReactor extends Reactor implements ShouldQueue
41+
{
42+
public function onBookingCancelled(BookingCancelled $event, StoredEvent $storedEvent)
43+
{
44+
$user = User::query()->findOrFail($event->userId);
45+
Notification::send($user, new BookingCancelledNotification('订单取消'));
46+
}
47+
```
48+
49+
## 源码分析
50+
当我们执行如下代码：
51+
```php
52+
BookingRoot::retrieve($bookingAggregateId)->cancel()->persist();
53+
```
54+
* ##### `Spatie\EventSourcing\AggregateRoots\AggregateRoot` 的 persist 方法（将事件保存到数据库）：
55+
```php
56+
public function persist()
57+
{
58+
$storedEvents = $this->persistWithoutApplyingToEventHandlers();
59+
60+
$storedEvents->each(fn (StoredEvent $storedEvent) => $storedEvent->handleForAggregateRoot());
61+
62+
$this->aggregateVersionAfterReconstitution = $this->aggregateVersion;
63+
64+
return $this;
65+
}
66+
67+
protected function persistWithoutApplyingToEventHandlers(): LazyCollection
68+
{
69+
$this->ensureNoOtherEventsHaveBeenPersisted();
70+
71+
$storedEvents = $this
72+
->getStoredEventRepository()
73+
->persistMany(
74+
$this->getAndClearRecordedEvents(),
75+
$this->uuid(),
76+
$this->aggregateVersion,
77+
);
78+
79+
return $storedEvents;
80+
}
81+
```
82+
在代码 `$this->persistWithoutApplyingToEventHandlers` 会调用 `Spatie\EventSourcing\StoredEvents\Repositories\EloquentStoredEventRepository` 的 persistMany 再循环要保存的事件通过 persist 保存到 `Spatie\EventSourcing\StoredEvents\Models\EloquentStoredEvent` 最后存到数据库表 stored_events 中。`Spatie\EventSourcing\StoredEvents\Repositories\EloquentStoredEventRepository` 代码如下:
83+
```php
84+
class EloquentStoredEventRepository implements StoredEventRepository
85+
{
86+
public function persistMany(array $events, string $uuid = null, int $aggregateVersion = null): LazyCollection
87+
{
88+
$storedEvents = [];
89+
foreach ($events as $event) {
90+
$storedEvents[] = $this->persist($event, $uuid, $aggregateVersion);
91+
}
92+
93+
return new LazyCollection($storedEvents);
94+
}
95+
96+
public function persist(ShouldBeStored $event, string $uuid = null, int $aggregateVersion = null): StoredEvent
97+
{
98+
/** @var EloquentStoredEvent $eloquentStoredEvent */
99+
$eloquentStoredEvent = new $this->storedEventModel();
100+
101+
$eloquentStoredEvent->setOriginalEvent($event);
102+
103+
$eloquentStoredEvent->setRawAttributes([
104+
'event_properties' => app(EventSerializer::class)->serialize(clone $event),
105+
'aggregate_uuid' => $uuid,
106+
'aggregate_version' => $aggregateVersion,
107+
'event_class' => $this->getEventClass(get_class($event)),
108+
'meta_data' => json_encode($event->metaData()),
109+
'created_at' => Carbon::now(),
110+
]);
111+
112+
$eloquentStoredEvent->save();
113+
114+
return $eloquentStoredEvent->toStoredEvent();
115+
}
116+
```
117+
* ##### 事件监听代码逻辑：
118+
```php
119+
$storedEvents = $this->persistWithoutApplyingToEventHandlers();
120+
```
121+
返回的是 item 为 `Spatie\EventSourcing\StoredEvents\StoredEvent` 的 `LazyCollection`
122+
第二步：
123+
```php
124+
$storedEvents->each(fn (StoredEvent $storedEvent) => $storedEvent->handleForAggregateRoot());
125+
```
126+
循环调用每一个 `Spatie\EventSourcing\StoredEvents\StoredEvent` 的 `handleForAggregateRoot` 方法。
127+
```php
128+
namespace Spatie\EventSourcing\StoredEvents;
129+
class StoredEvent implements Arrayable
130+
{
131+
132+
public function handleForAggregateRoot(): void
133+
{
134+
$this->handle();
135+
136+
if (! config('event-sourcing.dispatch_events_from_aggregate_roots', false)) {
137+
return;
138+
}
139+
140+
$this->event->firedFromAggregateRoot = true;
141+
event($this->event);
142+
}
143+
144+
public function handle()
145+
{
146+
// Projector 和 Reactor 如果是同步 那么在 Projectionist 的 handleWithSyncEventHandlers 方法处理
147+
Projectionist::handleWithSyncEventHandlers($this);
148+
149+
if (method_exists($this->event, 'tags')) {
150+
$tags = $this->event->tags();
151+
}
152+
153+
if (! $this->shouldDispatchJob()) {
154+
return;
155+
}
156+
$storedEventJob = call_user_func(
157+
[config('event-sourcing.stored_event_job'), 'createForEvent'],
158+
$this,
159+
$tags ?? []
160+
);
161+
// Projector 和 Reactor 如果是异步就把 Spatie\EventSourcing\StoredEvents\HandleStoredEventJob 推送到队列
162+
dispatch($storedEventJob->onQueue($this->getQueueName()));
163+
}
164+
```
165+
上面代码中的 $this->event 为 `App\Bookings\Event\BookingCancelled`
166+
$storedEventJob 为 `Spatie\EventSourcing\StoredEvents\HandleStoredEventJob` 属性 $storedEvent 为代码中的 $this。
167+
168+
* ##### Projector 和 Reactor 同步监听逻辑
169+
上面代码中同步调用代码：
170+
```php
171+
Projectionist::handleWithSyncEventHandlers($this);
172+
```
173+
我们找到 `Spatie\EventSourcing\Projectionist` 的 handleWithSyncEventHandlers
174+
```php
175+
namespace Spatie\EventSourcing;
176+
use Spatie\EventSourcing\EventHandlers\EventHandlerCollection;
177+
class Projectionist
178+
{
179+
private EventHandlerCollection $projectors;
180+
181+
private EventHandlerCollection $reactors;
182+
183+
private bool $catchExceptions;
184+
185+
private bool $replayChunkSize;
186+
187+
private bool $isProjecting = false;
188+
189+
private bool $isReplaying = false;
190+
191+
public function __construct(array $config)
192+
{
193+
$this->projectors = new EventHandlerCollection();
194+
$this->reactors = new EventHandlerCollection();
195+
196+
$this->catchExceptions = $config['catch_exceptions'];
197+
$this->replayChunkSize = $config['replay_chunk_size'];
198+
}
199+
public function handleWithSyncEventHandlers(StoredEvent $storedEvent): void
200+
{
201+
$projectors = $this->projectors
202+
->forEvent($storedEvent)
203+
->syncEventHandlers();
204+
205+
$this->applyStoredEventToProjectors($storedEvent, $projectors);
206+
207+
$reactors = $this->reactors
208+
->forEvent($storedEvent)
209+
->syncEventHandlers();
210+
211+
$this->applyStoredEventToReactors($storedEvent, $reactors);
212+
}
213+
}
214+
```
215+
上面代码中的 `handleWithSyncEventHandlers ` 方法其实是通过 `EventHandlerCollection` 筛选出同步的 Projector 和 Reactor。
216+
如果是 Projector 执行 `$this->applyStoredEventToProjectors($storedEvent, $projectors)`，
217+
如果是 Reactor 执行 `$this->applyStoredEventToReactors(StoredEvent $storedEvent, Collection $reactors):`
218+
```php
219+
private function applyStoredEventToProjectors(StoredEvent $storedEvent, Collection $projectors): void
220+
{
221+
$this->isProjecting = true;
222+
223+
foreach ($projectors as $projector) {
224+
$this->callEventHandler($projector, $storedEvent);
225+
}
226+
227+
$this->isProjecting = false;
228+
}
229+
230+
private function applyStoredEventToReactors(StoredEvent $storedEvent, Collection $reactors): void
231+
{
232+
foreach ($reactors as $reactor) {
233+
$this->callEventHandler($reactor, $storedEvent);
234+
}
235+
}
236+
237+
private function callEventHandler(EventHandler $eventHandler, StoredEvent $storedEvent): bool
238+
{
239+
try {
240+
$eventHandler->handle($storedEvent);
241+
} catch (Exception $exception) {
242+
if (! $this->catchExceptions) {
243+
throw $exception;
244+
}
245+
246+
$eventHandler->handleException($exception);
247+
248+
event(new EventHandlerFailedHandlingEvent($eventHandler, $storedEvent, $exception));
249+
250+
return false;
251+
}
252+
253+
return true;
254+
}
255+
```
256+
在上面的案例中到了这一步,我们触发的是 `$this->applyStoredEventToProjectors($storedEvent, $projectors)`。
257+
当调用 `$this->callEventHandler($projector, $storedEvent)` 时候：
258+
$projector 就是 `App\Bookings\Projector\BookingsProjector` ，
259+
$storedEvent 就是 `Spatie\EventSourcing\StoredEvents\StoredEvent` 它的属性 event 是 `App\Bookings\Event\BookingCancelled`。
260+
所以 `callEventHandler` 里面的 `$eventHandler->handle($storedEvent)` 可以理解为：`$projector->handle($storedEvent)`;
261+
其实就是调用了 `App\Bookings\Projector\BookingsProjector` 的 handle 方法。然后 handle 执行了`onBookingCancelled ` 方法。
262+
263+
* #### Projector 和 Reactor 异步监听逻辑
264+
上面代码中异步调用代码：
265+
```php
266+
dispatch($storedEventJob->onQueue($this->getQueueName()));
267+
```
268+
Laravel 的监听最后通过 `Illuminate\Bus\Dispatcher` 执行方法 `dispatchNow($command, $handler = null)` 还是会调用 $storedEventJob 的 handler 方法。
269+
我们找到 `Spatie\EventSourcing\StoredEvents\HandleStoredEventJob` 的相关代码：
270+
```php
271+
namespace Spatie\EventSourcing\StoredEvents;
272+
class HandleStoredEventJob implements HandleDomainEventJob, ShouldQueue
273+
{
274+
use InteractsWithQueue, Queueable, SerializesModels;
275+
276+
public StoredEvent $storedEvent;
277+
278+
public array $tags;
279+
280+
public function __construct(StoredEvent $storedEvent, array $tags)
281+
{
282+
$this->storedEvent = $storedEvent;
283+
284+
$this->tags = $tags;
285+
}
286+
287+
public function handle(Projectionist $projectionist): void
288+
{
289+
$projectionist->handle($this->storedEvent);
290+
}
291+
}
292+
```
293+
`handle` 继续调用了 `Spatie\EventSourcing\Projectionist` 的 `handle` 方法,代码如下：
294+
```php
295+
namespace Spatie\EventSourcing;
296+
class Projectionist
297+
{
298+
public function handle(StoredEvent $storedEvent): void
299+
{
300+
$projectors = $this->projectors
301+
->forEvent($storedEvent)
302+
->asyncEventHandlers();
303+
304+
$this->applyStoredEventToProjectors(
305+
$storedEvent,
306+
$projectors
307+
);
308+
309+
$reactors = $this->reactors
310+
->forEvent($storedEvent)
311+
->asyncEventHandlers();
312+
313+
$this->applyStoredEventToReactors(
314+
$storedEvent,
315+
$reactors
316+
);
317+
}
318+
}
319+
```
320+
到这里其实和同步的逻辑就一样了。 `handle` 和 `handleWithSyncEventHandlers` 的区别只是前者筛选的是异步的 Projector 和 Reactor。后者筛选的是同步的。然后走到
321+
`$this->applyStoredEventToProjectors($storedEvent, $projectors)` 以及
322+
`$this->applyStoredEventToReactors(StoredEvent $storedEvent, Collection $reactors)`。
323+
案例中：触发的就是`App\Bookings\Reactor\NotificationReactor` 的 `handle` 方法。然后 `handle` 执行了 `onBookingCancelled` 方法。
324+
325+
326+
## 流程图
327+
![](http://www.plantuml.com/plantuml/png/dPFVQXD15CRlzodE2_G5Sb4eKX5Ha5JmmjniaY4PsSwiixFYUXMjs8O6RRMAD2eKAfPYRK2KDD7qPNPcurjuajc4dUsc_hbPTcU_-SwPttT6KkaHqDyU9qVRgZjogloXuxj2qXhrNIPXfT4GfE5AKkPSMdzMFNu_94okIIv8VVK1lfQ9pmEAtv6bp2YizLk2toCrIJcZWVtdcilg7idikywh3c5rcBJdkA7aB5ol4k7SNLgsEeII-lag7cp7m-_W4n5CZ2t11JsCUnl9tX5KMAg_GsMJXtB5zxs8iiPjFct0T2I2lDkb5ERcgVLDbqMWDf-fmqtLm-T-t3yspSRdxzN9M-TIjpyKmwEeqN7o_5ItFuqFEhEQW9N-fOPrFZp0-PxgVW0goJh4_K4sIqIMx3-56-wZw0htF9DadazMNpBz6IRwz4NSRo60d6Lp2leg5vQHVdEclxvsCbBBUWxQx8QBSbXQjiODQQNVN81wsO4oSStxJaUVV4owkshdyw_MS3pQhJ13r7XFncCjOZLxAYplNAcIRe_KLa-tM_fTszZZiCsA1nMMbgwmR7xgomRIdV608W15jmHAQVmrGP0TSZJthaYRIsS1ZnzFvZommwsk6WxIu8eyKNA6LoyCce1d1foqa2meNjp-T0Vut5zW3rayDfpYCldOKpxsictqzsaQE9ES_Y_Gtm00)
328+
329+
## 注意点
330+
[Laravel-Event-Sourcing](https://spatie.be/docs/laravel-event-sourcing/v4/introduction) 异步处理机制是判断监听当前事件的 Projector 和 Reactor 是否 `instanceof ShouldQueue`，如果比对有一个则将当前事件创建的 HandleStoredEventJob 推送到 queue。无论同步还是异步的执行。都是拿到监听这个事件的所有 Projector 和 Reactor。如案例中的 BookingCancelled，则循环执行所有 Projector 和 Reactor 的 onBookingCancelled 方法。代码如下：
331+
```php
332+
private function applyStoredEventToProjectors(StoredEvent $storedEvent, Collection $projectors): void
333+
{
334+
$this->isProjecting = true;
335+
336+
foreach ($projectors as $projector) {
337+
$this->callEventHandler($projector, $storedEvent);
338+
}
339+
340+
$this->isProjecting = false;
341+
}
342+
343+
private function applyStoredEventToReactors(StoredEvent $storedEvent, Collection $reactors): void
344+
{
345+
foreach ($reactors as $reactor) {
346+
$this->callEventHandler($reactor, $storedEvent);
347+
}
348+
}
349+
350+
private function callEventHandler(EventHandler $eventHandler, StoredEvent $storedEvent): bool
351+
{
352+
try {
353+
$eventHandler->handle($storedEvent);
354+
} catch (Exception $exception) {
355+
if (! $this->catchExceptions) {
356+
throw $exception;
357+
}
358+
359+
$eventHandler->handleException($exception);
360+
361+
event(new EventHandlerFailedHandlingEvent($eventHandler, $storedEvent, $exception));
362+
363+
return false;
364+
}
365+
366+
return true;
367+
}
368+
```
369+
如果我们不希望循环执行所有 Projector 和 Reactor 时候被中断,可以在 event-sourcing.php 配置文件的 catch_exceptions 设置为 true,然后在 Projector 和 Reactor 自定义 handleException 方法处理我们的异常。

0 commit comments

Comments
 (0)