本地事件总线和事务

热情之风

本地事件总线和事务

通过重写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. }

复制代码

分布式事件总线和事务

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

来源:https://www.cnblogs.com/donpangpang/Undeclared/17939853
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！