.net core 消息流處理流程

前言

2020年即將進入尾聲,分享一下在現公司業務處理流程,一塊兒討論在分佈式場景下,如何經過消息流的方式處理各類複雜的業務場景,這裏涉及到一些經常使用組件,後面結合場景與代碼來具體說明

場景說明

這裏就拿我負責的短信應用來舉例,它由3個核心模塊組成

1. 短信網關(接收客戶提交短信,接收通道短信回執,轉發回執給客戶...)
2. 計費服務(短信計費,購買套餐...)
3. 短信通道(短信提交到通道...)

痛點

• 批量提交營銷類短信,如客戶一次提交10w-20w條短信
• 短信實時計費,延遲問題
• 通道限流控制
• 通知回執延時問題

組件

mysql
redis
阿里雲日誌

流程以下

痛點解決方案

• 批量提交短信

批量提交這裏主要如下幾種狀況

1. 短信內容一致,手機號多個
   
2. 短信內容不一致(例如內容攜帶用戶信息等... 一般採用excel上傳)
   
3. 循環調用接口,提交短信
   

這裏僅針對狀況3來講明,首先將用戶信息存入redis(先讀緩存,再讀庫),減小在驗證與鑑權時對數據庫的查詢壓力,校驗經過的消息開始寫入收單隊列,並記錄日誌(注意日誌必定要異步寫入),隊列使用的redis隊列,以目前的業務承載能力,是徹底沒有問題的,收單接口的qps能夠經過分佈式來提高,這個得益於k8s容器伸縮,平時咱們通常是5個pod在運行(至關於負載5個應用),在節假日高峯期能夠起10個pod,這裏性能的瓶頸主要集中在redis隊列,若是有更高要求能夠嘗試換成rabbitmq,kafka等

• mysql批量持久化

這也一直是我比較迷的一個地方,數據庫使用的阿里雲mysql,單條循環插入速度在200/s左右,我這裏採用的dapper,經過拼接values來批量插入,速度大概能達到3000/s,後面看看有沒有更好的方案

• 短信實時計費,延遲問題

在分佈式狀況下,實時計費又是一個比較大的性能瓶頸,直接讀庫,並修改用戶條數顯然是不行的,這樣會出現髒讀,致使最終數據不許確而出損(程序員就要背鍋),並且效率低下,使用redis分佈式鎖等同於將分佈式改爲單應用,須要頻繁更新緩存裏的用戶短信餘量,速度大概在150/s - 180/s,消費速度過慢會致使隊列裏數據堆積,短信延遲太高,特別是通知類短信(如獲取驗證碼)對延遲有較高的要求,這裏使用redis的計數器來實現,經過計數器遞減的方式,若是短信餘量<0則用戶短信餘量不足,再單獨起一個任務,每分鐘同步一下用戶短信餘量,用戶充值與短信失敗回退,也是對計數器的操做

• 通道限流控制

通道限流主要是供貨商對通道進行流量限制,超頻的短信會直接失敗,通常出如今營銷類短信,由於每條通道的價格(與地域,三網,到達率...有關)都不同,每一個用戶都會分配通道,爲了讓短信儘可能成功,程序須要進行限流,這裏也是使用redis計數器實現,設置一個1分鐘失效的緩存,超過頻次後,會嘗試其它通道,沒有可用通道則再次寫入隊列

• 通知回執延時問題

系統拿到回執後,須要將回執通知給客戶配置的http地址,並獲得客戶的響應,未響應的回執會再次入列,直到客戶響應,或者超過推送策略(推送3次或超過多長時間),有些客戶配置的http地址響應很是慢,或者乾脆是一些訪問超時的地址,這樣會致使通知延遲,通知類髮卡密業務的短信對回執有較高要求,這裏經過多線程來實現,經過建立多個線程從隊列裏獲取回執並轉發

List<Task> tasks = new List<Task>();
  for (int i = 0; i < 10; i++)
  {
      tasks.Add(Task.Run(async () =>{...}));
  }
  await Task.WhenAll(tasks);

• 注意事項

1. 由於操做redis的地方很是多,爲了便於管理,全部redis的key建議統一寫在常量類裏,並寫上註釋,並制定對應的規範,方便維護

/// 項目名_類型_操做_參數    有效期
project_queue_action_params

2. 業務日誌,業務日誌方便排查問題,建議不要偷懶,在業務的入口跟出口都寫上對應的日誌信息

• 推薦我本身寫的一個Redis消息隊列中間件InitQ,操做簡單能夠下載的玩玩
  https://github.com/wmowm/initq