Hangfire項目實踐分享
目錄
項目中使用Hangfire已經快一年了，期間經歷過很多次的試錯及升級優化，才達到現在的穩定效果。趁最近不是太忙，自己在github上做了個案列，也是拿來跟大家分享下，案例是從項目里剝離出來的，有興趣的可以訪問 這里.
什么是Hangfire
Hangfire 是一個開源的.NET任務調度框架，目前1.6+版本已支持.NET Core。個人認為它最大特點在於內置提供集成化的控制台,方便后台查看及監控：
另外，Hangfire包含三大核心組件：客戶端、持久化存儲、服務端，官方的流程介紹圖如下：
從圖中可以看出，這三個核心組件是可以分離出來單獨部署的，例如可以部署多台Hangfire服務，提高處理后台任務的吞吐量。關於任務持久化存儲，支持Sqlserver，MongoDb，Mysql或是Redis等等。
Hangfire基礎
基於隊列的任務處理(Fire-and-forget jobs)
基於隊列的任務處理是Hangfire中最常用的，客戶端使用BackgroundJob類的靜態方法Enqueue來調用，傳入指定的方法（或是匿名函數），Job Queue等參數.
var jobId = BackgroundJob.Enqueue(
() => Console.WriteLine("Fire-and-forget!"));
在任務被持久化到數據庫之后，Hangfire服務端立即從數據庫獲取相關任務並裝載到相應的Job Queue下，在沒有異常的情況下僅處理一次，若發生異常，提供重試機制，異常及重試信息都會被記錄到數據庫中，通過Hangfire控制面板可以查看到這些信息。
延遲任務執行(Delayed jobs)
延遲（計划）任務跟隊列任務相似，客戶端調用時需要指定在一定時間間隔后調用：
var jobId = BackgroundJob.Schedule(
() => Console.WriteLine("Delayed!"),
TimeSpan.FromDays(7));
定時任務執行(Recurring jobs)
定時（循環）任務代表可以重復性執行多次，支持CRON表達式：
RecurringJob.AddOrUpdate(
() => Console.WriteLine("Recurring!"),
Cron.Daily);
延續性任務執行(Continuations)
延續性任務類似於.NET中的Task,可以在第一個任務執行完之后緊接着再次執行另外的任務：
BackgroundJob.ContinueWith(
jobId,
() => Console.WriteLine("Continuation!"));
其實還有批量任務處理，批量任務延續性處理(Batch Continuations)，但這個需要商業授權及收費。在我看來，官方提供的開源版本已經基本夠用。
與quartz.net對比
在項目沒有引入Hangfire之前，一直使用的是Quartz.net。個人認為Quartz.net在定時任務處理方面優勢如下：
- 支持秒級單位的定時任務處理，但是Hangfire只能支持分鍾及以上的定時任務處理
原因在於Hangfire用的是開源的NCrontab組件，跟linux上的crontab指令相似。
-
更加復雜的觸發器，日歷以及任務調度處理
-
可配置的定時任務
但是為什么要換Hangfire? 很大的原因在於項目需要一個后台可監控的應用，不用每次都要從服務器拉取日志查看，在沒有ELK的時候相當不方便。Hangfire控制面板不僅提供監控，也可以手動的觸發執行定時任務。如果在定時任務處理方面沒有很高的要求，比如一定要5s定時執行，Hangfire值得擁有。拋開這些，Hangfire優勢太明顯了：
-
持久化保存任務、隊列、統計信息
-
重試機制
-
多語言支持
-
支持任務取消
-
支持按指定
Job Queue處理任務 -
服務器端工作線程可控，即job執行並發數控制
-
分布式部署，支持高可用
-
良好的擴展性，如支持IOC、Hangfire Dashboard授權控制、Asp.net Core、持久化存儲等
說了這么多的優點，我們可以有個案例，例如秒殺場景：用戶下單->訂單生成->扣減庫存，Hangfire對於這種分布式的應用處理也是適用的，最后會給出實現。
Hangfire擴展
重點說一下上面提到的第8點，Hangfire擴展性，大家可以參考 這里，有幾個擴展是很實用的.
Hangfire Dashborad日志查看
Hangfire.Console提供類似於console-like的日志體驗，與Hangfire dashboard集成：
用法如下：
public void SimpleJob(PerformContext context)
{
context.WriteLine($"{DateTime.Now.ToString("yyyy/MM/dd HH:mm:ss")} SimpleJob Running ...");
var progressBar = context.WriteProgressBar();
foreach (var i in Enumerable.Range(1, 50).ToList().WithProgress(progressBar))
{
System.Threading.Thread.Sleep(1000);
}
}
不僅支持日志輸入到控制面板，也支持在線進度條展示.
Hangfire Dashborad授權
Hangfire.Dashboard.Authorization這個擴展應該都能理解，給Hangfire Dashboard
提供授權機制，僅授權的用戶才能訪問。其中提供兩種授權機制：
- OWIN-based authentication
- Basic authentication
可以參考提供案例 ，我實現的是基本認證授權:
var options = new DashboardOptions
{
AppPath = HangfireSettings.Instance.AppWebSite,
AuthorizationFilters = new[]
{
new BasicAuthAuthorizationFilter ( new BasicAuthAuthorizationFilterOptions
{
SslRedirect = false,
RequireSsl = false,
LoginCaseSensitive = true,
Users = new[]
{
new BasicAuthAuthorizationUser
{
Login = HangfireSettings.Instance.LoginUser,
// Password as plain text
PasswordClear = HangfireSettings.Instance.LoginPwd
}
}
} )
}
};
app.UseHangfireDashboard("", options);
IOC容器之Autofac
Hangfire對於每一個任務(Job)假如都寫在一個類里，然后使用BackgroundJob/RecurringJob對方法(實例或靜態)進行調用，這樣會導致模塊間太多耦合。實際項目中，依賴倒置原則可以降低模塊之間的耦合性，Hangfire也提供了IOC擴展，其本質是重寫JobActivator類。
Hangfire.Autofac是官方提供的開源擴展，用法參考如下：
GlobalConfiguration.Configuration.UseAutofacActivator(container);
RecurringJob擴展
關於RecurringJob定時任務，我寫了一個擴展 RecurringJobExtensions,在使用上做了一下增強，具體有兩點：
使用特性RecurringJobAttribute發現定時任務
public class RecurringJobService
{
[RecurringJob("*/1 * * * *")]
[DisplayName("InstanceTestJob")]
[Queue("jobs")]
public void InstanceTestJob(PerformContext context)
{
context.WriteLine($"{DateTime.Now.ToString("yyyy/MM/dd HH:mm:ss")} InstanceTestJob Running ...");
}
[RecurringJob("*/5 * * * *")]
[DisplayName("JobStaticTest")]
[Queue("jobs")]
public static void StaticTestJob(PerformContext context)
{
context.WriteLine($"{DateTime.Now.ToString("yyyy/MM/dd HH:mm:ss")} StaticTestJob Running ...");
}
}
使用json配置文件注冊定時任務
[AutomaticRetry(Attempts = 0)]
[DisableConcurrentExecution(90)]
public class LongRunningJob : IRecurringJob
{
public void Execute(PerformContext context)
{
context.WriteLine($"{DateTime.Now.ToString("yyyy/MM/dd HH:mm:ss")} LongRunningJob Running ...");
var runningTimes = context.GetJobData<int>("RunningTimes");
context.WriteLine($"get job data parameter-> RunningTimes: {runningTimes}");
var progressBar = context.WriteProgressBar();
foreach (var i in Enumerable.Range(1, runningTimes).ToList().WithProgress(progressBar))
{
Thread.Sleep(1000);
}
}
}
Json配置文件如下：
[{
"job-name": "Long Running Job",
"job-type": "Hangfire.Samples.LongRunningJob, Hangfire.Samples",
"cron-expression": "*/2 * * * *",
"job-data": {
"RunningTimes": 300
}
}]
實現接口IRecurringJob來定義具體的定時任務，這樣的寫法與Quartz.net相似，可以很方便的實現Quartz.net到Hangfire的遷移。類似地，參考了quartz.net，
使用job-data-map這樣的方式來定義整個任務執行期間的上下文有狀態的job.
var runningTimes = context.GetJobData<int>("RunningTimes");
詳細用法可以直接參考項目文檔。
與MSMQ集成
Hangfire server在處理每個job時，會將job先裝載到事先定義好的job queue中，比如一次性加載1000個job,在默認的sqlsever實現中是直接將這些job queue中的
job id儲存到數據庫中，然后再取出執行。大量的job會造成任務的延遲性執行，所以更有效的方式是將任務直接加載到MSMQ中。
實際應用中，MSMQ隊列不存在時一定要手工創建，而且必須是事務性的隊列，權限也要設置，用法如下:
public static IGlobalConfiguration<SqlServerStorage> UseMsmq(this IGlobalConfiguration<SqlServerStorage> configuration, string pathPattern, params string[] queues)
{
if (string.IsNullOrEmpty(pathPattern)) throw new ArgumentNullException(nameof(pathPattern));
if (queues == null) throw new ArgumentNullException(nameof(queues));
foreach (var queueName in queues)
{
var path = string.Format(pathPattern, queueName);
if (!MessageQueue.Exists(path))
using (var queue = MessageQueue.Create(path, transactional: true))
queue.SetPermissions("Everyone", MessageQueueAccessRights.FullControl);
}
return configuration.UseMsmqQueues(pathPattern, queues);
}
持久化存儲之Redis
Hangfire中定義的job存儲到sqlserver不是性能最好的選擇，使用Redis存儲，性能將會是巨大提升(下圖來源於Hangfire.Pro.Redis).
Hangfire.Pro提供了基於servicestack.redis的redis擴展組件，然而商業收費，不開源。
但是，有另外的基於StackExchange.Redis的開源實現 Hangfire.Redis.StackExchange，
github上一直在維護，支持.NET Core，項目實測穩定可用. 該擴展相當簡單：
services.AddHangfire(x =>
{
var connectionString = Configuration.GetConnectionString("hangfire.redis");
x.UseRedisStorage(connectionString);
});
Hangfire最佳實踐
配置最大job並發處理數
Hangfire server在啟動時會初始化一個最大Job處理並發數量的閾值，系統默認為20,可以根據服務器配置設置並發處理數。最大閾值的定義除了考慮服務器配置以外，
也需要考慮數據庫的最大連接數，定義太多的並發處理數量可能會在同一時間耗盡數據連接池。
app.UseHangfireServer(new BackgroundJobServerOptions
{
//wait all jobs performed when BackgroundJobServer shutdown.
ShutdownTimeout = TimeSpan.FromMinutes(30),
Queues = queues,
WorkerCount = Math.Max(Environment.ProcessorCount, 20)
});
使用 DisplayNameAttribute特性構造缺省的JobName
public interface IOrderService : IAppService
{
/// <summary>
/// Creating order from product.
/// </summary>
/// <param name="productId"></param>
[AutomaticRetry(Attempts = 3)]
[DisplayName("Creating order from product, productId:{0}")]
[Queue("apis")]
void CreateOrder(int productId);
}
目前netstandard暫不支持缺省的jobname,因為需要單獨引用組件System.ComponentModel.Primitives,hangfire官方給出的答復是盡量保證少的Hangfire.Core組件的依賴。
Hangfire在調用Background/RecurringJob創建job時應盡量使傳入的參數簡單.
Hangfire job中參數（包括參數值）及方法名都序列化為json持久化到數據庫中，所以參數應盡量簡單，如傳入單據ID，這樣才不會使Job Storage呈爆炸性增長。
為Hangfire客戶端調用定義統一的REST APIs
定義統一的REST APIs可以規范並集中管理整個項目的hangfire客戶端調用，同時避免到處引用hangfire組件。使用例如Swagger這樣的組件來給不同的應用方(Consumer)提供文檔幫助，應用方可以是App,Webservice,Microservices等。
/// <summary>
/// Creating order from product.
/// </summary>
/// <param name="productId"></param>
/// <returns></returns>
[Route("create")]
[HttpPost]
public IActionResult Create([FromBody]string productId)
{
if (string.IsNullOrEmpty(productId))
return BadRequest();
var jobId = BackgroundJob.Enqueue<IOrderService>(x => x.CreateOrder(productId));
BackgroundJob.ContinueWith<IInventoryService>(jobId, x => x.Reduce(productId));
return Ok(new { Status = 1, Message = $"Enqueued successfully, ProductId->{productId}" });
}
利用Topshelf + Owin Host將hangfire server 宿主到Windows Service.
不推薦將hangfire server 宿主到如ASP.NET application 中，需要有一堆配置。個人喜好問題，推薦將hangfire server 單獨部署到windows service, 利用Topshelf+Owin Host：
/// <summary>
/// OWIN host
/// </summary>
public class Bootstrap : ServiceControl
{
private static readonly ILog _logger = LogProvider.For<Bootstrap>();
private IDisposable webApp;
public string Address { get; set; }
public bool Start(HostControl hostControl)
{
try
{
webApp = WebApp.Start<Startup>(Address);
return true;
}
catch (Exception ex)
{
_logger.ErrorException("Topshelf starting occured errors.", ex);
return false;
}
}
public bool Stop(HostControl hostControl)
{
try
{
webApp?.Dispose();
return true;
}
catch (Exception ex)
{
_logger.ErrorException($"Topshelf stopping occured errors.", ex);
return false;
}
}
}
日志配置
從Hangfire 1.3.0開始，Hangfire引入了日志組件LibLog,所以應用不需要做任何改動就可以兼容如下日志組件：
-
Serilog
-
NLog
-
Log4Net
-
EntLib Logging
-
Loupe
-
Elmah
例如，配置 serilog如下，LibLog組件會自動發現並使用serilog
Log.Logger = new LoggerConfiguration()
.MinimumLevel.Verbose()
.WriteTo.LiterateConsole()
.WriteTo.RollingFile("logs\\log-{Date}.txt")
.CreateLogger();
Hangfire多實例部署（高可用）
下圖是一個多實例Hangfire服務部署：
其中，關於Hangfire Server Node 節點可以根據實際需要水平擴展.
上述提到過一個秒殺場景：用戶下單->訂單生成->扣減庫存，實現參考github項目Hangfire.Topshelf.
HF.Samples.Consumer
服務應用消費方(App/Webservice/Microservices等。)
HF.Samples.APIs
統一的REST APIs管理
HF.Samples.Console
Hangfire 控制面板
HF.Samples.ServerNode
Hangfire server node cli 工具，使用如下：
@echo off
set dir="cluster"
dotnet run -p %dir%\HF.Samples.ServerNode nodeA -q order -w 100
dotnet run -p %dir%\HF.Samples.ServerNode nodeB -q storage -w 100
上述腳本為創建兩個Hangfire server nodeA, nodeB分別用來處理訂單、倉儲服務。
-q 指定hangfire server 需要處理的隊列，-w表示Hangfire server 並發處理job數量。
可以為每個job queue創建一個hangfire實例來處理更多的job.
