相信絕大部分開發者都接觸過用戶註冊的流程，通常情況下大概的流程如下所示：
- 接收用戶提交註冊信息
- 持久化註冊信息（數據庫+redis）
- 發送註冊成功短信（郵件）
- 寫操作日誌（可選）
偽代碼如下：
public async Task<IActionResult> Reg([FromBody] User user)
{
_logger.LogInformation("持久化數據開始");
await Task.Delay(50);
_logger.LogInformation("持久化結束");
_logger.LogInformation("發送短信開始");
await Task.Delay(100);
_logger.LogInformation("發送短信結束");
_logger.LogInformation("操作日誌開始");
await _logRepository.Insert(new Log { Txt = "註冊日誌" });
_logger.LogInformation("操作日誌結束");
return Ok("註冊成功");
}
在以上的代碼中，我使用Task.Delay方法阻塞主線程，用以模擬實際場景中的執行耗時。以上流程應該是包含了絕大部分註冊流程所需要的操作。對於任何開發者來講，以上業務流程沒任何難度，無非是順序的執行各個流程的代碼即可。
稍微有點開發經驗的應該會將以上的流程進行拆分，但有些人可能就要問了，為什麼要拆分呢？拆分之後的代碼應該怎麼寫呢？下面我們就來簡單聊下如此場景的正確打開方式。
首先，註冊成功的依據應該是是否成功的將用戶信息持久化（至於是先持久化到數據庫，異或是先寫到redis不在本篇文章討論的範疇），至於發送註冊短信（郵件）以及寫日誌的操作應該不能成為影響註冊是否成功的因素，而發送短信/郵件等相關操作通常情況下也是比較耗時的，所以在對此接口做性能優化時，可優先考慮將短信/郵件以及寫日誌等相關操作與主流程（持久化數據）拆分，使其不阻塞主流程的執行，從而達到提高響應速度的目的。
知道了為什麼要拆，但具體如何拆分呢？怎樣才能用最少的改動，達到所需的目的呢？
條條大路通羅馬，所以要達成我們的目的也是有很多方案的，具體選擇哪種方案需要根據具體的業務場景，業務體量等多種因素綜合考慮，下面我將一一介紹分析相關方案。
在正式介紹可用方案前，筆者想先介紹一種很多新手容易錯誤使用的一種方案（因為筆者就曾經天真的使用過這種錯誤的方案）。
提到異步，絕大部分.net開發者應該第一想到的就是Task，async，await等，的確，async，await的語法糖簡化了.net開發者異步編程的門檻，減少了很多代碼量。通常一個返回Task類型的方法，在被調用時，會在方法的前面加上await，表示需要等待此方法的執行結果，再繼續執行後面的代碼。但如果不加await時，則不會等待方法的執行結果，進而也不會阻塞主線程。所以，有些人可能就會將發送短信/郵件以及寫日誌的操作如下方式進行改造。
public async Task<IActionResult> Reg1([FromBody] User user)
{
_logger.LogInformation("持久化數據開始");
await Task.Delay(50);
_logger.LogInformation("持久化結束");
_ = Task.Run(async () =>
{
_logger.LogInformation("發送短信開始");
await Task.Delay(100);
_logger.LogInformation("發送短信結束");
_logger.LogInformation("操作日誌開始");
await _logRepository.Insert(new Log { Txt = "註冊日誌" });
_logger.LogInformation("操作日誌結束");
});
return Ok("註冊成功");
}
然後使用jmeter分別壓測改造前和改造后的接口，結果如下：
有沒有被驚訝到？就這樣一個簡單的改造，吞吐量就提高了三四倍。既然已經提高了三四倍，那為什麼說這是一種錯誤的改造方法嗎？各位看官且往下看。
熟悉.netcore的大佬，應該都知道.netcore的依賴注入的生命周期吧。通常情況下，注入的生命周期包括：Singleton，Scope，Transient。
在以上的流程中，假如寫操作日誌的實例的生命周期是Scope，當在Task中調用Controller獲取到的實例的方法時，因為Task.Run並沒有阻塞主線程，當調用Action return后，當前請求的scope注入的對象會被回收，如果對象被回收之後，Task.Run還未執行完，則會報System.ObjectDisposedException: Cannot access a disposed object. 異常。意思是，不能訪問一個已disposed的對象。正確的做法是使用IServiceScopeFactory創建一個新的作用域，在新的作用域中獲取獲取日誌倉儲服務的實例。這樣就可以避免System.ObjectDisposedException異常了。
改造后的示例代碼如下：
public async Task<IActionResult> Reg1([FromBody] User user)
{
_logger.LogInformation("持久化數據開始");
await Task.Delay(50);
_logger.LogInformation("持久化結束");
_ = Task.Run(async () =>
{
using (var scope = _scopeFactory.CreateScope())
{
var sp = scope.ServiceProvider;
var logRepository = sp.GetService<ILogRepository>();
_logger.LogInformation("發送短信開始");
await Task.Delay(100);
_logger.LogInformation("發送短信結束");
_logger.LogInformation("操作日誌開始");
await logRepository.Insert(new Log { Txt = "註冊日誌" });
_logger.LogInformation("操作日誌結束");
}
});
return Ok("註冊成功");
}
雖然得到了正解，但上述的代碼着實有點多，如果一個項目有多個相似的業務場景，就要考慮對CreateScope相關的操作進行封裝。
下面就來一一介紹下筆者覺得實現此業務場景的幾種方案。
1.消息隊列
2.Quartz任務調度組件
3.Hangfire任務調度組件
4.Weshare.TransferJob(推薦)
首先說下消息隊列的方式。準確的說，消息隊列應該是這種場景的最優解決方案，消息隊列的其中一個比較重要的特性就是解耦，從而提高吞吐量。但並不是所有的應用程序都需要上消息隊列。有些業務場景使用消息隊列時，往往會給人一種”殺雞用牛刀”的感覺。
其次Quartz和Hangfire都是任務調度框架，都提供了可實現以上業務場景的邏輯，但Quartz和Hangfire都需要持久化作業數據。雖然Hangfire提供了內存版本，但經過我的測試，發現Hangfire的內存版本特別消耗內存，所以不太推薦使用任務調度框架來實現類似於這樣的業務邏輯。
最後，也就是本文的重點，筆者結合了消息隊列和任務調度的思想，實現了一個輕量級的轉移作業到後台執行的組件。此組件完美的解決了Scope生命周期實例獲取的問題，一行代碼將不需要等待的操作轉移到後台線程執行。
接入步驟如下：
1.使用nuget安裝Weshare.TransferJob
2.在Stratup中注入服務。
services.AddTransferJob();
3.通過構造函數或其他方法獲取到IBackgroundRunService的實例。
4.調用實例的Transfer方法將作業轉移到後台線程。
_backgroundRunService.Transfer(log=>log.Insert(new Log(){Txt = "註冊日誌"}));
就是這麼簡單的實現了這樣的業務場景，不僅簡化了代碼，而且大大提高了系統的吞吐量。
下面再來一起分析下Weshare.TransferJob的核心代碼（畢竟文章要點題）。各位器宇不凡的看官請繼續往下看。
下面的代碼是AddTransferJob方法的實現：
public static IServiceCollection AddTransferJob(this IServiceCollection services)
{
services.AddSingleton<IBackgroundRunService, BackgroundRunService>();
services.AddHostedService<TransferJobHostedService>();
return services;
}
聰明”絕頂”的各位看官應該已經發現上述代碼的關鍵所在。是的， 你沒有看錯，此組件的就是利用.net core提供的HostedService在後台執行被轉移的作業的。
我們再來一起看看TransferJobHostedService的代碼：
public class TransferJobHostedService:BackgroundService
{
private IBackgroundRunService _runService;
public TransferJobHostedService(IBackgroundRunService runService)
{
_runService = runService;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
await _runService.Execute(stoppingToken);
}
}
}
這個類的代碼也很簡單，重寫了BackgroundService類的ExecuteAsync，循環調用IBackgroundRunService實例的Execute方法。所以，最最關鍵的代碼是IBackgroundRunService的實現類中。
詳細代碼如下：
public class BackgroundRunService : IBackgroundRunService
{
private readonly SemaphoreSlim _slim;
private readonly ConcurrentQueue<LambdaExpression> queue;
private ILogger<BackgroundRunService> _logger;
private readonly IServiceProvider _serviceProvider;
public BackgroundRunService(ILogger<BackgroundRunService> logger, IServiceProvider serviceProvider)
{
_slim = new SemaphoreSlim(1);
_logger = logger;
_serviceProvider = serviceProvider;
queue = new ConcurrentQueue<LambdaExpression>();
}
public async Task Execute(CancellationToken cancellationToken)
{
try
{
await _slim.WaitAsync(cancellationToken);
if (queue.TryDequeue(out var job))
{
using (var scope = _serviceProvider.GetRequiredService<IServiceScopeFactory>().CreateScope())
{
var action = job.Compile();
var isTask = action.Method.ReturnType == typeof(Task);
var parameters = job.Parameters;
var pars = new List<object>();
if (parameters.Any())
{
var type = parameters[0].Type;
var param = scope.ServiceProvider.GetRequiredService(type);
pars.Add(param);
}
if (isTask)
{
await (Task)action.DynamicInvoke(pars.ToArray());
}
else
{
action.DynamicInvoke(pars.ToArray());
}
}
}
}
catch (Exception e)
{
_logger.LogError(e.ToString());
}
}
public void Transfer<T>(Expression<Func<T, Task>> expression)
{
queue.Enqueue(expression);
_slim.Release();
}
public void Transfer(Expression<Action> expression)
{
queue.Enqueue(expression);
_slim.Release();
}
}
納尼？嫌代碼多看不懂？那咱們一起來剖析下吧。
首先，此類有三個較重要的私有變量，對應的類型分別是SemaphoreSlim， ConcurrentQueue ，IServiceProvider。
其中SemaphoreSlim是為了控制後台作業執行的順序的，在構造函數中初始化了此對象的信號量為1，表示在後台服務的ExecuteAsync方法的循環中每次只能有一個作業執行。
ConcurrentQueue 的對象是用來存儲被轉移到後台服務執行的作業的邏輯，所以使用LambdaExpression作為隊列的類型。
IServiceProvider是為了解決依賴注入的生命周期的。
然後在Execute方法中，第一行代碼如下：
await _slim.WaitAsync(cancellationToken);
作用是等待一個信號量，當沒有可用的信號量時，會阻塞線程的執行，這樣在後台服務的ExecuteAsync方法的死循環就不會一直執行下去，只有獲取到信號量才會繼續執行。
當獲取到信號量后，則說明有新的作業等待執行，所以此時則需要從隊列中讀出要執行的LambdaExpression表達式，創建一個新的Scope后，編譯此表達式樹，判斷返回類型，獲取泛型的具體類型，最後獲取到泛型對應的實例，執行對應的方法。
另外，Transfer方法就是暴露給調用者的方法，用於將表達式樹寫到隊列中，同時釋放信號量。
到此為止，Weshare.TransferJob的實現原理已分析完畢，由於此組件的原理只是將任務轉移到後台進行執行，所以並不是適合對事務有要求的場景。正如本文開頭所假設的場景，TransferJob最適合的場景還是那些和主操作關聯性較低的、失敗或成功並不會影響業務的正常運行。
同時，此組件的定位就是小而美，像延遲執行、定時執行的功能在最初的規劃中其實是有的，後來發現這些功能quartz已經有了，所以沒必要重複造這樣的輪子。
後期會根據使用場景，嘗試加入異常重試機制，以及異常通知回調機制。
最後，不知道有沒有較真的看官想計算下代碼量是否超過120行。
為了證明我不是標題黨，現將此組件進行開源，地址是：
https://github.com/fuluteam/WeShare.TransferJob
橋豆麻袋，筆者辛苦敲的代碼，難道各位看官想白嫖嗎? 點個贊再走唄。點完贊還有力氣的話，如果git上能點個star的話，那也是最好不過的。小生這廂先行謝過。
本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
【其他文章推薦】
※教你寫出一流的銷售文案?
※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益
※回頭車貨運收費標準
※別再煩惱如何寫文案,掌握八大原則!
※超省錢租車方案
※產品缺大量曝光嗎?你需要的是一流包裝設計!
