使用EventBus + Redis釋出訂閱模式提升業務執行效能
阿新 • • 發佈：2020-03-03
前言
最近一直奔波於面試，面了幾家公司的研發。有讓我受益頗多的面試經驗，也有讓我感覺浪費時間的面試經歷~
因為疫情原因，最近宅在家裡也沒事，就想著使用Redis配合事件匯流排去實現下具體的業務。
-
需求
一個簡單的電商，有幾個重要的需求點
商品下單後TODO
- 儲存訂單資訊
- 鎖定商品庫存
- 訊息推送商家端
訂單支付後TODO
- 儲存訂單支付資訊
- 商品庫存減少
- 訊息推送商家端
- 會員積分調整
技術思路
這裡用控制檯實現上面的業務功能外，自行編寫一個基於C#反射特性的事件匯流排，方便具體業務事件的後續擴充套件，比如訂單支付後後續還要加會員訊息推送啥的。使用Redis的釋出訂閱模式對事件處理進行非同步化，提升執行效能。
完成事件匯流排
這裡先不急著將上面的訂單、支付、會員 等進行建模。先將事件匯流排的架子搭好。首先需要理解事件匯流排在業務系統的目的是什麼。
事件匯流排存在目的最重要的就是解耦 。我們需要實現的效果就是針對指定事件源物件觸發事件後，但凡註冊了該事件引數的事件處理類則開始執行相關程式碼。
下圖可以看出我們的事件處理類均需要引用事件引數，所有事件處理類都是基於對事件引數處理的需求上來的。
但是！並不是意味建立了事件處理類就一定會去執行！能否執行除了取決於事件源的觸發外就是必須有一層註冊(也可稱對映)。
在WinForm程式裡處處可見事件的繫結，如 this.button1.OnClick+=button1OnClick；
那麼在這裡我將繫結事件放置到一個字典裡。C#的字典Dictionary是個key value的鍵值對資料集合，鍵和值都可以是任意資料型別。
我們可以將事件處理類EventHandle和事件引數EventData作為鍵和值儲存到字典裡。在事件源觸發時根據EventData反向找出所有的EventHandle
思路就是這樣，開始編碼了。
定義事件引數介面，後續具體業務的事件引數介面均要繼承它。
/// <summary>
/// 事件引數介面
/// </summary>
public interface IEventData
{
/// <summary>
/// 事件源物件
/// </summary>
object Source { get; set; }
///// <summary>
///// 事件發生的資料
///// </summary>
//TDataModel Data { get; set; }
/// <summary>
/// 事件發生時間
/// </summary>
DateTime Time { get; set; }
}
需要一個事件處理介面，後續具體業務的事件處理介面均需要繼承它
/// <summary>
/// 事件實現介面
/// </summary>
public interface IEventHandle<T> where T : IEventData
{
/// <summary>
/// 處理等級
/// 方便事件匯流排觸發時候可以有序的執行相應
/// </summary>
/// <returns></returns>
int ExecuteLevel { get; }
/// <summary>
/// 事件執行
/// </summary>
/// <param name="eventData">事件引數</param>
void Execute(T eventData);
}
現在已經將事件引數和事件處理都抽象出來了，接下來是要實現上面說的註冊容器的實現了。
/// <summary>
/// 事件倉庫
/// </summary>
public interface IEventStore
{
/// <summary>
/// 事件註冊
/// </summary>
/// <param name="handle">事件實現物件</param>
/// <param name="data">事件引數</param>
void EventRegister(Type handle, Type data);
/// <summary>
/// 事件取消註冊
/// </summary>
/// <param name="handle">事件實現物件</param>
void EventUnRegister(Type handle);
/// <summary>
/// 獲取事件處理物件
/// </summary>
/// <param name="data"></param>
/// <returns></returns>
Type GetEventHandle(Type data);
/// <summary>
/// 根據事件引數獲取事件處理集合
/// </summary>
/// <typeparam name="TEventData">事件引數型別</typeparam>
/// <param name="data">事件引數</param>
/// <returns></returns>
IEnumerable<Type> GetEventHandleList<TEventData>(TEventData data);
}
實現上面的介面
/// <summary>
/// 基於反射實現的事件倉儲
/// 儲存事件處理物件和事件引數
/// </summary>
public class ReflectEventStore : IEventStore
{
private static Dictionary<Type, Type> StoreLst;
public ReflectEventStore()
{
StoreLst = new Dictionary<Type, Type>();
}
public void EventRegister(Type handle, Type data)
{
if (handle == null || data == null) throw new NullReferenceException();
if (StoreLst.Keys.Contains(handle))
throw new ArgumentException($"事件匯流排已註冊型別為{nameof(handle)} !");
if (!StoreLst.TryAdd(handle, data))
throw new Exception($"註冊{nameof(handle)}型別到事件匯流排失敗！");
}
public void EventUnRegister(Type handle)
{
if (handle == null) throw new NullReferenceException();
StoreLst.Remove(handle);
}
public Type GetEventHandle(Type data)
{
if (data == null) throw new NullReferenceException();
Type handle = StoreLst.FirstOrDefault(p => p.Value == data).Key;
return handle;
}
public IEnumerable<Type> GetEventHandleList<TEventData>(TEventData data)
{
if (data == null) throw new NullReferenceException();
var items = StoreLst.Where(p => p.Value == data.GetType())
.Select(k => k.Key);
return items;
}
}
根據上面程式碼可以看出來，我們儲存到Dictionary內的是Type型別，GetEventHandleList方法最終獲取的是一個List<Type>的集合。
我們需要在下面建立的EventBus類裡迴圈List<Type>並且執行這個事件處理類的Execute方法。
實現EventBus
/// <summary>
/// 事件匯流排服務
/// </summary>
public class EventBus : ReflectEventStore
{
public void Trigger<TEventData>(TEventData data, SortType sort = SortType.Asc) where TEventData : IEventData
{
// 這裡如需保證順序執行則必須迴圈兩次 - -....
var items = GetEventHandleList(data).ToList();
Dictionary<object, Tuple<Type, int>> ds = new Dictionary<object, Tuple<Type, int>>();
foreach (var item in items)
{
var instance = Activator.CreateInstance(item);
MethodInfo method = item.GetMethod("get_ExecuteLevel");
int value = (int)method.Invoke(instance, null);
ds.Add(instance, new Tuple<Type, int>(item, value));
}
var lst = sort == SortType.Asc ? ds.OrderBy(p => p.Value.Item2).ToList() : ds.OrderByDescending(p => p.Value.Item2).ToList();
foreach (var k in lst)
{
MethodInfo method = k.Value.Item1.GetMethod("Execute");
method.Invoke(k.Key, new object[] { data });
}
}
}
上面可以看到，我們的事件匯流排是支援對繫結的事件處理物件進行有序處理，需要依賴下面這個列舉
/// <summary>
/// 排序型別
/// </summary>
public enum SortType
{
/// <summary>
/// 升序
/// </summary>
Asc = 1,
/// <summary>
/// 降序
/// </summary>
Desc = 2
}
好了，至此，我們的簡易版的事件匯流排就出來了~ 接下來就是去建模、實現相應的事件引數和事件處理類了。
建立訂單模型：
/// <summary>
/// 訂單模型
/// </summary>
public class OrderModel
{
/// <summary>
/// 訂單ID
/// </summary>
public Guid Id { get; set; }
/// <summary>
/// 使用者ID
/// </summary>
public Guid UserId { get; set; }
/// <summary>
/// 訂單建立時間
/// </summary>
public DateTime CreateTime { get; set; }
/// <summary>
/// 商品名稱
/// </summary>
public string ProductName { get; set; }
/// <summary>
/// 購買數量
/// </summary>
public int Number { get; set; }
/// <summary>
/// 訂單金額
/// </summary>
public decimal Money { get; set; }
}
建立訂單下單事件引數
public interface IOrderCreateEventData : IEventData
{
/// <summary>
/// 訂單資訊
/// </summary>
OrderModel Order { get; set; }
}
/// <summary>
/// 訂單建立事件引數
/// </summary>
public class OrderCreateEventData : IOrderCreateEventData
{
public OrderModel Order { get; set; }
public object Source { get; set; }
public DateTime Time { get; set; }
}
OK~接下來就是實現我們上面需求上的那些功能了。
- 儲存訂單資訊
- 鎖定商品庫存
- 訊息推送商家端
這裡我不實現儲存訂單資訊的事件處理物件，我預設此業務必須同步處理，至於後面兩個則可以採取非同步處理。通過下面程式碼建立相應的事件處理類。
訂單建立事件之訊息推送商家端處理類。
/// <summary>
/// 訂單建立事件之訊息處理類
/// </summary>
public class OrderCreateEventNotifyHandle : IEventHandle<IOrderCreateEventData>
{
public int ExecuteLevel { get; private set; }
public OrderCreateEventNotifyHandle()
{
Console.WriteLine($"建立OrderCreateEventNotifyHandle物件");
this.ExecuteLevel = 2;
}
public void Execute(IOrderCreateEventData eventData)
{
Thread.Sleep(1000);
Console.WriteLine($"執行訂單建立事件之訊息推送!訂單ID:{eventData.Order.Id.ToString()},商品名稱:{eventData.Order.ProductName}");
}
}
訂單建立訊息之鎖定庫存處理類
/// <summary>
/// 訂單建立事件 鎖定庫存 處理類
/// </summary>
public class OrderCreateEventStockLockHandle : IEventHandle<IOrderCreateEventData>
{
public int ExecuteLevel { get; private set; }
public OrderCreateEventStockLockHandle()
{
Console.WriteLine($"建立OrderCreateEventStockLockHandle物件");
this.ExecuteLevel = 1;
}
public void Execute(IOrderCreateEventData eventData)
{
Thread.Sleep(1000);
Console.WriteLine($"執行訂單建立事件之庫存鎖定!訂單ID:{eventData.Order.Id.ToString()},商品名稱:{eventData.Order.ProductName}");
}
}
OK~ 到main方法下開始執行訂單建立相關程式碼。
static void Main(string[] args)
{
Guid userId = Guid.NewGuid();
Stopwatch stopwatch = new Stopwatch();
stopwatch.Start();
EventBus eventBus = new EventBus();
eventBus.EventRegister(typeof(OrderCreateEventNotifyHandle), typeof(OrderCreateEventData));
eventBus.EventRegister(typeof(OrderCreateEventStockLockHandle), typeof(OrderCreateEventData));
var order = new Order.OrderModel()
{
CreateTime = DateTime.Now,
Id = Guid.NewGuid(),
Money = (decimal)300.00,
Number = 1,
ProductName = "鮮花一束",
UserId = userId
};
Console.WriteLine($"模擬儲存訂單");
Thread.Sleep(1000);
eventBus.Trigger(new OrderCreateEventData()
{
Order = order
});
stopwatch.Stop();
Console.WriteLine($"下單總耗時:{stopwatch.ElapsedMilliseconds}毫秒");
Console.ReadLine();
}
至此，我們採取事件匯流排的方式成功將需求實現了，執行後結果如下:
可以看到我們的下單總耗時是3038毫秒，如您所見，我們解決了程式碼的耦合性但是沒有解決程式碼的執行效率。
下一章，將我們的Redis的釋出訂閱模式再加入進來，看是否能改善我們的程式碼執行效率~~
原始碼在下一篇部落格上提供下載地址(畢竟現在才完成一半~)