ToB企服应用市场:ToB评测及商务社交产业平台

标题: 本地事件总线和事务 [打印本页]

---

作者: 水军大提督    时间: 2024-3-16 17:48
标题: 本地事件总线和事务
本地事件总线和事务

通过重写Ef Core的SaveChanges/SaveChangesAsync来实现事务。当然，如果您愿意实现仓储层，可以在仓储层实现展开对应实体包含的事件，并且调整事件的处理顺序。
Github仓库地址：soda-event-bus
实现AggregateRoot类

AggregateRoot类主要通过一个集合来记录本次事务的所有事件，到保存前再展开读取，在Abp中采用的ICollection记录的本地事件，通过实现一个排序器来保证顺序问题，我这里直接采用了ConcurrentQueue，保证原子操作的同时保证了顺序性，实现更简单一些。

1. public abstract class AggregateRoot
2. {
3.     public ConcurrentQueue<object> LocalEvents { get; } = new();
5.     public void AddLocalEvent<TEvent>(TEvent eventData) where TEvent : IEvent
6.     {
7.         LocalEvents.Enqueue(eventData);
8.     }
10.     public bool GetLocalEvent(out object? @event)
11.     {
12.         LocalEvents.TryDequeue(out var eventData);
14.         @event = eventData;
15.         return @event is not null;
16.     }
18.     public void ClearLocalEvents()
19.     {
20.         LocalEvents.Clear();
21.     }
22. }

复制代码

重写DbContext

主要是从ServiceProvider中获取对应实体类包含的事件，并且找到对应的Handler进行处理，然后再当作一个事务提交。

1. public class EventBusDbContext<TDbContext> : DbContext
2.     where TDbContext : DbContext
3. {
4.     private readonly IServiceProvider _serviceProvider;
6.     public EventBusDbContext(DbContextOptions<TDbContext> options, IServiceProvider serviceProvider) : base(options)
7.     {
8.         _serviceProvider = serviceProvider;
9.     }
11.     public override int SaveChanges()
12.     {
13.         return base.SaveChanges();
14.     }
16.     public override int SaveChanges(bool acceptAllChangesOnSuccess)
17.     {
18.         return base.SaveChanges(acceptAllChangesOnSuccess);
19.     }
21.     public override async Task<int> SaveChangesAsync(CancellationToken cancellationToken = default)
22.     {
23.         await HandleEventsAsync();
25.         return await base.SaveChangesAsync(cancellationToken);
26.     }
28.     public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = default)
29.     {
30.         await HandleEventsAsync();
32.         return await base.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
33.     }
35.     private async Task HandleEventsAsync()
36.     {
37.         foreach (var entityEntry in ChangeTracker.Entries<AggregateRoot>())
38.         {
39.             while (entityEntry.Entity.GetLocalEvent(out var @event))
40.             {
41.                 if (@event is null) break;
43.                 await HandleEventAsync(@event);
44.             }
46.             entityEntry.Entity.ClearLocalEvents();
47.         }
48.     }
50.     private async Task HandleEventAsync(object @event)
51.     {
52.         var eventHandlerType = typeof(IAsyncEventHandler<>).MakeGenericType(@event.GetType());
53.         var eventHandler = _serviceProvider.GetRequiredService(eventHandlerType);
55.         var method = eventHandler.GetType().GetMethod(nameof(IAsyncEventHandler<IEvent>.HandleAsync));
56.         var exceptionHandleMethod = eventHandlerType.GetMethod(nameof(IAsyncEventHandler<IEvent>.HandleException));
58.         try
59.         {
60.             await (Task)method!.Invoke(eventHandler, new[] { @event })!;
61.         }
62.         catch (Exception ex)
63.         {
64.             exceptionHandleMethod!.Invoke(eventHandler, new[] { @event, ex });
65.         }
66.     }
67. }

复制代码

分布式事件总线和事务

根据需要扩展即可，基本逻辑相同，但可能需要增加确认机制等。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4