背景

事件发布/订阅是一种非常强大的模式，它可以帮助业务组件间实现完全解耦，不同的业务组件只依赖事件，只关注哪些事件是需要自己处理的，而不用关注谁来处理自己发布事件，事件追溯(Event Sourcing)也是基于事件发布/订阅的。在微服务架构中，事件发布/订阅有非常多的应用场景。今天我给大家分享一个基于ASP.NET Core的单体程序使用事件发布/订阅的例子，针对分布式项目的事件发布/订阅比较复杂，难点是事务处理，后续我会另写一篇博文来演示。

案例说明

当前我们有一个基于ASP.NET Core的电子商务系统，在项目的初期，业务非常简单，只有一个购物车模块和一个订单模块，所有的代码都放在一个项目中。

整个项目使用了一个简单的三层架构。

这里当用户提交购物车的时候，程序会在

ShoppingCartManager

SubmitShoppingCart

• 修改当前购物车的状态为完成
• 根据购物车中的物品创建一个新订单
• 给用户发邮件

代码如下：

public void SubmitShoppingCart(string shoppingCartId)
{
var shoppingCart = _unitOfWork.ShoppingCartRepository
.GetShoppingCart(shoppingCartId);

_unitOfWork.ShoppingCartRepository
.SubmitShoppingCart(shoppingCartId);

_unitOfWork.OrderRepository
.CreatOrder(new CreateOrderDTO
{
Items = shoppingCart.Items
.Select(p => new NewOrderItemDTO
{
ItemId = p.ItemId,
Name = p.Name,
Price = p.Price
}).ToList()
});

//这里为了简化代码，我用命令行表示发送邮件的逻辑
Console.WriteLine("Confirm Email Sent.");

_unitOfWork.Save();
}

根据SOLID设计原则中的单一责任原则，如果一个类承担的职责过多，就等于把这些职责耦合在一起了。这里生成订单和发送邮件都不应该是当前

SubmitShoppingCart

新的架构图

以下是使用事件发布/订阅之后的系统架构图。

• 这里我们会创建一个购物车提交事件

  ShoppingCartSubmittedEvent

  。
• 当站点启动的时候，我们会在一个名为

  EventHandlerContainer

  的类中注册订阅

  ShoppingCartSubmittedEvent

  事件的2个处理类

  CreateOrderHandler

  和

  ConfirmEmailSentHandler

  。
• 在

  SubmitShoppingCart

  方法中，我们会做2件事情： 更改当前购物车的状态。
• 发布

  ShoppingCartSubmittedEvent

  事件。

CreateOrderHandler

OrderManager

ConfirmEmailSentHandler

好的，下面让我们来一步一步实现以上描述的代码。

添加事件基类

这里我们首先定义一个事件基类，其中暂时只添加了一个属性

OccuredOn

public class EventBase
{
public EventBase()
{
OccuredOn = DateTime.Now;
}

protected DateTime OccuredOn
{
get;
set;
}
}

定义购物车提交事件

接下来我们就需要创建购物车提交事件类

ShoppingCartSubmittedEvent

EventBase

public class ShoppingCartSubmittedEvent : EventBase
{
public ShoppingCartSubmittedEvent()
{
Items = new List<ShoppingCartSubmittedItem>();
}

public List<ShoppingCartSubmittedItem> Items { get; set; }
}

public class ShoppingCartSubmittedItem
{
public string ItemId { get; set; }

public string Name { get; set; }

public decimal Price { get; set; }

}

定义事件处理器接口

为了添加事件处理器，我们首先需要定义一个泛型接口类

IEventHandler

public interface IEventHandler<T> where T : EventBase
{
void Run(T obj);

Task RunAsync(T obj);
}

这个泛型接口类的是泛型类型必须继承自

EventBase

Run

RunAsync

为购物车提交事件编写事件处理器

有了事件处理器接口，接下来我们就可以开始为购物车提交事件添加事件处理器了。这里我们为了实现前面定义的逻辑，我们需要创建2个处理器

CreateOrderHandler

ConfirmEmailSentHandler

CreateOrderHandler.cs

public class CreateOrderHandler : IEventHandler<ShoppingCartSubmittedEvent>
{
private IOrderManager _orderManager = null;

public CreateOrderHandler(IOrderManager orderManager)
{
_orderManager = orderManager;
}

public void Run(ShoppingCartSubmittedEvent obj)
{
_orderManager.CreateNewOrder(new Models.DTOs.CreateOrderDTO
{
Items = obj.Items.Select(p => new Models.DTOs.NewOrderItemDTO
{
ItemId = p.ItemId,
Name = p.Name,
Price = p.Price
}).ToList()
});
}

public Task RunAsync(ShoppingCartSubmittedEvent obj)
{
return Task.Run(() =>
{
Run(obj);
});
}
}

ConfirmEmailSentHandler.cs

public class ConfirmEmailSentHandler : IEventHandler<ShoppingCartSubmittedEvent>
{
public void Run(ShoppingCartSubmittedEvent obj)
{
Console.WriteLine("Confirm Email Sent.");
}

public Task RunAsync(ShoppingCartSubmittedEvent obj)
{
return Task.Run(() =>
{
Console.WriteLine("Confirm Email Sent.");
});
}
}

为

OrderManager

类添加创建订单方法

IOrderManager.cs

public interface IOrderManager
{
string CreateNewOrder(CreateOrderDTO dto);
}

OrderManager.cs

public class OrderManager : IOrderManager
{
private IOrderRepository _orderRepository;

public OrderManager(IOrderRepository orderRepository)
{
_orderRepository = orderRepository;
}

public string CreateNewOrder(CreateOrderDTO dto)
{
var orderId = _orderRepository.CreatOrder(dto);

Console.WriteLine($"One order created: {JsonConvert.SerializeObject(dto)}");

return orderId;
}
}

创建

EventHandlerContainer

下面我们来编写最核心的事件处理器容器。在这里我们的事件处理器容器完成了3个功能

• 订阅事件(Subscribe Event)
• 取消订阅事件(Unsubscribe Event)
• 发布事件(Publish Event)

public class EventHandlerContainer
{
private IServiceProvider _serviceProvider = null;
private static Dictionary<string, List<Type>> _mappings = new Dictionary<string, List<Type>>();

public EventHandlerContainer(IServiceProvider serviceProvider)
{
_serviceProvider = serviceProvider;
}

public static void Subscribe<T, THandler>()
where T : EventBase
where THandler : IEventHandler<T>
{
var name = typeof(T).Name;

if (!_mappings.ContainsKey(name))
{
_mappings.Add(name, new List<Type> { });
}

_mappings[name].Add(typeof(THandler));
}

public static void Unsubscribe<T, THandler>()
where T : EventBase
where THandler : IEventHandler<T>
{
var name = typeof(T).Name;
_mappings[name].Remove(typeof(THandler));

if (_mappings[name].Count == 0)
{
_mappings.Remove(name);
}
}

public void Publish<T>(T o) where T : EventBase
{
var name = typeof(T).Name;

if (_mappings.ContainsKey(name))
{
foreach (var handler in _mappings[name])
{
var service = (IEventHandler<T>)_serviceProvider.GetService(handler);

service.Run(o);
}
}
}

public async Task PublishAsync<T>(T o) where T : EventBase
{
var name = typeof(T).Name;

if (_mappings.ContainsKey(name))
{
foreach (var handler in _mappings[name])
{
var service = (IEventHandler<T>)_serviceProvider.GetService(handler);

await service.RunAsync(o);
}
}
}
}

在程序启动时，注册事件订阅

现在我们来

Startup.cs

ConfigureServices

public void ConfigureServices(IServiceCollection services)
{
services.AddMvc().SetCompatibilityVersion(CompatibilityVersion.Version_2_2);

services.AddScoped<IOrderManager, OrderManager>();
services.AddScoped<IShoppingCartManager, ShoppingCartManager>();
services.AddScoped<IShoppingCartRepository, ShoppingCartRepository>();
services.AddScoped<IOrderRepository, OrderRepository>();
services.AddScoped<IUnitOfWork, UnitOfWork>();
services.AddScoped<CreateOrderHandler>();
services.AddScoped<ConfirmEmailSentHandler>();
services.AddScoped<EventHandlerContainer>();

EventHandlerContainer.Subscribe<ShoppingCartSubmittedEvent, CreateOrderHandler>();
EventHandlerContainer.Subscribe<ShoppingCartSubmittedEvent, ConfirmEmailSentHandler>();
}

Scoped

IUnitOfWork

Save

修改ShoppingCartManager

最后我们来修改

ShoppingCartManager

public void SubmitShoppingCart(string shoppingCartId)
{
var shoppingCart = _unitOfWork.ShoppingCartRepository
.GetShoppingCart(shoppingCartId);

_unitOfWork.ShoppingCartRepository
.SubmitShoppingCart(shoppingCartId);

_container.Publish(new ShoppingCartSubmittedEvent()
{
Items = shoppingCart
.Items
.Select(p => new ShoppingCartSubmittedItem
{
ItemId = p.ItemId,
Name = p.Name,
Price = p.Price
})
.ToList()
});

_unitOfWork.Save();
}

这样

ShoppingCartManager

最终效果

现在让我们启动项目，

首先我们使用[POST] /api/shoppingCarts来添加一个新的购物车, 这个API会返回当前购物车的Id

然后我们使用[PUT] /api/shoppingCarts/ShoppingCart_636872897140555966来模拟提交购物车，程序返回操作成功

最后我们查看一下控制台的输出日志

2个事件处理器都被正确触发了。

总结

至此我们的代码重构完成。 最终的代码中，

SubmitShoppingCart

这样的方式的好处不仅仅是完成了代码的解耦，针对后续的扩展也非常有利，想想一下，如果在未来当前项目需求追加这样一个功能，当提交购物车的时候，除了要发送确认邮件，还要发送手机短信。这时候你根本不需要去修改

ShoppingCartManager

ShoppingCartSubmittedEvent