🪴 Colo's Blog

Search

SearchSearch

最新文章

See 18 more →

最新笔记

See 12 more →

中介者模式的事件通知物化数据

May 10, 2024, 5 min read

最近想到利用设计模式中的中介者实现一个异步的物化数据的方法

首先，你接收到一个请求比如“创建一个订单（同一个商品可以买多个）”，创建完订单后，还要查询购买的各个{商品-总价}，我们假设计算{商品-总价}列表的时间很长，那体验并不好。

所以可以运用在创建玩订单后发送中介者事件通知一个异步的请求（不阻塞创建订单的请求），在后台计算比如说{Id,List{ProductId,TotalPrice}}的数据，这个Id是根据请求名和订单Id生成的唯一值，List{ProductId,TotalPrice}则转化成json字符串与Id一起存储在数据库中，当根据订单Id查询List{ProductId,TotalPrice}时，直接将计算好的json字符串返回。不过因为本质是缓存，就必须考虑原子事务，以下是两个问题以及解决方法。

Q：假如创建完订单后立刻查看总价列表而后台还没物化好数据怎么办？

A：现在数据库中寻找订单Id假如不存在就直接根据原始数据计算，这是不可避免的耗时。

Q：是否会出现查看总价列表时后台还没物化好数据所以查到了上次的记录

A：会，所以在创建完订单时同时需要删除之前物化的数据保持一致性。

以下是chatgpt生成的dotnet实现例子

1.定义中介者和相关事件

首先，我们需要一个中介者来协调各种操作，如订单创建后的异步任务。我们可以使用 .NET 的 MediatR 库来实现中介者模式。

// 定义一个中介者事件
public class OrderCreatedEvent : INotification
{
    public int OrderId { get; set; }
    public List<OrderItem> Items { get; set; }
}

public class OrderItem
{
    public int ProductId { get; set; }
    public int Quantity { get; set; }
}
2. 创建订单逻辑

在订单创建逻辑中，我们将触发一个中介者事件，这样就可以在不影响主线程的情况下进行异步计算。

public class OrderService
{
    private readonly IMediator _mediator;

    public OrderService(IMediator mediator)
    {
        _mediator = mediator;
    }

    public async Task CreateOrderAsync(List<OrderItem> items)
    {
        // 此处逻辑为创建订单
        int orderId = 123; // 假设订单ID
        Task.Run(() => {
            _mediator.Publish(new OrderCreatedEvent { OrderId = orderId, Items = items });
        })


        // 返回创建成功的响应或视图
    }
}
3. 异步物化数据处理器

我们需要一个事件处理器来处理订单创建后的异步计算任务。

public class OrderCreatedEventHandler : INotificationHandler<OrderCreatedEvent>
{
    private readonly IDataStore _dataStore;
    private readonly PricingService _pricingService;

    public OrderCreatedEventHandler(IDataStore dataStore, PricingService pricingService)
    {
        _dataStore = dataStore;
        _pricingService = pricingService;
    }

    public async Task Handle(OrderCreatedEvent notification, CancellationToken cancellationToken)
    {
        var results = await _pricingService.CalculateTotalPricesAsync(notification.OrderId);
        string id = $"order-{notification.OrderId}";
        string json = JsonConvert.SerializeObject(results);
        await _dataStore.StoreMaterializedDataAsync(id, json);

        // 删除旧的物化数据
        await _dataStore.DeleteMaterializedDataAsync(id);
    }
}
4. 物化数据服务
public class PricingService
{
    private readonly IOrderRepository _orderRepository; // 假设这是获取订单数据的仓库接口

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

    public async Task<List<ProductTotalPrice>> CalculateTotalPricesAsync(int orderId)
    {
        var items = await _orderRepository.GetOrderItemsAsync(orderId);
        var results = new List<ProductTotalPrice>();
        foreach (var item in items)
        {
            var totalPrice = CalculateTotalPrice(item.ProductId, item.Quantity);
            results.Add(new ProductTotalPrice { ProductId = item.ProductId, TotalPrice = totalPrice });
        }
        return results;
    }

    private decimal CalculateTotalPrice(int productId, int quantity)
    {
        // 假设每个产品100元
        return 100m * quantity;
    }
}
5. 查询物化数据
public class DataService
{
    private readonly IDataStore _dataStore;
    private readonly PricingService _pricingService;

    public DataService(IDataStore dataStore, PricingService pricingService)
    {
        _dataStore = dataStore;
        _pricingService = pricingService;
    }

    public async Task<string> GetMaterializedDataAsync(int orderId)
    {
        string id = $"order-{orderId}";
        var data = await _dataStore.GetMaterializedDataAsync(id);
        if (data != null)
        {
            return data;
        }

        // 如果没有物化数据，则进行实时计算
        var calculatedData = await _pricingService.CalculateTotalPricesAsync(orderId);
        return JsonConvert.SerializeObject(calculatedData);
    }
}

评论