Hangfire只允許同時運行同一個任務

Hangfire有個機制可以確保所有任務都會被執行，如果當服務器停機了一段時間重新啟動時，在此期間的周期任務會幾乎同時執行。而大部分時候，我們希望同個周期任務每段時間只運行一個就行了。

或者是如果周期任務設置得過於頻繁，當之前的任務還沒執行完，我們也不希望繼續添加周期任務進隊列去排隊執行。

Hangfire有提供一個擴展https://docs.hangfire.io/en/latest/background-processing/throttling.html 

同個DisableConcurrentExecution我們可以限制同一個任務每次只會執行一個，但是如果有任務正在執行，這時候又有新任務過來，新任務並不會被刪除而是處於排隊狀態，等待前面的任務執行完。

 

 

而且，如果我們的任務用了同一個方法作為入口時（或者說我們需要根據方法的參數來確定是否為同一個任務），此時這個控制就不適用了。

參考了https://gist.github.com/sbosell/3831f5bb893b20e82c72467baf8aefea，我們可以用過濾器來實現，將運行期間進來的任務給取消掉。

代碼的具體實現為：

    /// <summary>
    /// 禁用多個排隊項目
    /// <remarks>同個任務取消並行執行，期間進來的任務不會等待，會被取消</remarks>
    /// </summary>
    public class DisableMultipleQueuedItemsFilter : JobFilterAttribute, IClientFilter, IServerFilter
    {
        private static readonly TimeSpan LockTimeout = TimeSpan.FromSeconds(5);
        private static readonly TimeSpan FingerprintTimeout = TimeSpan.FromHours(4);//任務執行超時時間

        public void OnCreating(CreatingContext filterContext)
        {
            var recurringJobId = filterContext.GetJobParameter<string>("RecurringJobId");
            if (!string.IsNullOrEmpty(recurringJobId)&&!AddFingerprintIfNotExists(filterContext.Connection, recurringJobId))
            {
                filterContext.Canceled = true;
            }
        }

        public void OnPerformed(PerformedContext filterContext)
        {
            var recurringJobId = filterContext.GetJobParameter<string>("RecurringJobId");
            if (!string.IsNullOrEmpty(recurringJobId))
            {
                RemoveFingerprint(filterContext.Connection, recurringJobId);
            }
        }

        private static bool AddFingerprintIfNotExists(IStorageConnection connection, string recurringJobId)
        {
            using (connection.AcquireDistributedLock(GetFingerprintLockKey(recurringJobId), LockTimeout))
            {
                var fingerprint = connection.GetAllEntriesFromHash(GetFingerprintKey(recurringJobId));

                if (fingerprint != null &&
                    fingerprint.ContainsKey("Timestamp") &&
                    DateTimeOffset.TryParse(fingerprint["Timestamp"], null, DateTimeStyles.RoundtripKind, out var timestamp) &&
                    DateTimeOffset.UtcNow <= timestamp.Add(FingerprintTimeout))
                {
                    // 有任務還未執行完，並且沒有超時
                    return false;
                }

                // 沒有任務執行，或者該任務已超時
                connection.SetRangeInHash(GetFingerprintKey(recurringJobId), new Dictionary<string, string>
            {
                { "Timestamp", DateTimeOffset.UtcNow.ToString("o") }
            });

                return true;
            }
        }

 

在OnCreating方法中，我們讀取RecurringJobId的值，獲取周期任務的id（同樣的id代表同一個周期任務），然后以這個id為key去設置一個超時。如果在此期間，如果拿到了key的值，以及設置的時間還未超時的話，我們通過設置filterContext.Canceled = true取消掉此任務。

 

 使用connection.AcquireDistributedLock在設置鍵值時添加分布式鎖，確保不會同時設置了多個相同的任務。使用connection.SetRangeInHash鍵RecurringJobId作為key，當前時間作為值保存。以此來確保在FingerprintTimeout的超時時間內，同個RecurringJobId的任務只能創建一個。或者等任務執行完后在OnPerformed方法中釋放掉這個鍵值。

在OnPerformed方法中，將我們在創建方法中設置的RecurringJobId key和對應的時間給刪除，這樣OnCreating可以繼續創建同一個RecurringJobId 的任務。

 

 

 

或者是普通觸發的任務，這時候沒有RecurringJobId 我們希望可以同個參數來控制，同樣的參數不能同時執行。我們可以通過這個方法來生成相應的key

        private static string GetFingerprint(Job job)
        {
            var parameters = string.Empty;
            if (job?.Arguments != null)
            {
                parameters = string.Join(".", job.Arguments);
            }
            if (job?.Type == null || job.Method == null)
            {
                return string.Empty;
            }
            var payload = $"{job.Type.FullName}.{job.Method.Name}.{parameters}";
            var hash = SHA256.Create().ComputeHash(System.Text.Encoding.UTF8.GetBytes(payload));
            var fingerprint = Convert.ToBase64String(hash);
            return fingerprint;
        }    

這樣我們就能確保我們希望的同一個任務不會同時在執行，而且周期任務也不會繼續在隊列中排隊

 

---

 

考慮到寫死鎖的key值不太合理，現添加特性來處理。

添加DisableMultipleInstanceAttribute特性，添加WebApiPullDisableMultipleInstance默認實現，取參數列表的第一個參數當做key

    /// <summary>
    /// 后台任務禁用重復任務排隊
    /// </summary>
    [AttributeUsage(AttributeTargets.Class)]
    public class DisableMultipleInstanceAttribute: BackgroundJobAttribute
    {
        private string _fingerprint;
        public DisableMultipleInstanceAttribute(string fingerprint = null)
        {
            _fingerprint = fingerprint;
        }

        public virtual string GetFingerprint(IReadOnlyList<object> methodArgs)
        {
            return _fingerprint;
        }

        public MultiTenancySides MultiTenancySides { get; set; } = MultiTenancySides.Tenant;
    }

    /// <summary>
    /// 接口禁用重復任務排隊
    /// </summary>
    public class WebApiPullDisableMultipleInstanceAttribute: DisableMultipleInstanceAttribute
    {
        public override string GetFingerprint(IReadOnlyList<object> methodArgs)
        {
            return methodArgs[0].ToString();
        }
    }

在基類中添加特性

 

 

 

添加hangfire的JobFilterAttributeFilterProvider的實現CustomJobAttributeFilterProvider，重寫GetTypeAttributes方法，添加我們新增的特性

    protected override IEnumerable<JobFilterAttribute> GetTypeAttributes(Job job)
        {
            foreach (var attribute in ReflectedAttributeCache.GetTypeFilterAttributes(job.Type))
            {
                if (attribute is CaptureContextAttribute)
                {
                    yield return new CaptureContextMessageAttribute();
                }

                if (attribute is AutomaticRetryAttribute automaticRetry)
                {
                    yield return new AutoRetryAttribute() { Attempts = automaticRetry.AutomaticRetry };
                }

                if (attribute is WebApiPullDisableMultipleInstanceAttribute apiPullDisable)
                {
                    yield return new DisableMultipleQueuedItemsAttribute(apiPullDisable);
                }
            }
        }

在DisableMultipleQueuedItemsAttribute的OnCreating中調用_disableMultipleInstanceAttribute.GetFingerprint獲取分布式鎖的key

    /// <summary>
    /// 禁用多個排隊項目
    /// <remarks>同個任務取消並行執行，期間進來的任務不會等待，會被取消</remarks>
    /// </summary>
    public class DisableMultipleQueuedItemsAttribute : JobFilterAttribute, IClientFilter, IServerFilter
    {
        private static readonly TimeSpan LockTimeout = TimeSpan.FromSeconds(5);
        private static readonly TimeSpan FingerprintTimeout = TimeSpan.FromHours(4);//任務執行超時時間

        private readonly DisableMultipleInstanceAttribute _disableMultipleInstanceAttribute;
        public DisableMultipleQueuedItemsAttribute(DisableMultipleInstanceAttribute attribute)
        {
            _disableMultipleInstanceAttribute = attribute;
        }

        public void OnCreating(CreatingContext filterContext)
        {
            var fingerprintKey = _disableMultipleInstanceAttribute.GetFingerprint(filterContext.Job.Args);
            if (string.IsNullOrEmpty(fingerprintKey))
            {
                throw new AppFatalExceptions("唯一鍵為空");
            }
            if (_disableMultipleInstanceAttribute.MultiTenancySides==MultiTenancySides.Tenant)
            {
                var contextMessage = filterContext.GetJobParameter<ContextMessage>("_ld_contextMessage");
                if (string.IsNullOrEmpty(contextMessage.TenantId))
                {
                    throw new AppFatalExceptions("租戶Id為空");
                }
                fingerprintKey = $"{contextMessage.TenantId}:{fingerprintKey}";
            }
            if (!AddFingerprintIfNotExists(filterContext.Connection, fingerprintKey))
            {
                filterContext.Canceled = true;
            }
        }

        public void OnPerformed(PerformedContext filterContext)
        {
            var fingerprintKey = _disableMultipleInstanceAttribute.GetFingerprint(filterContext.BackgroundJob.Job.Args);
            if (_disableMultipleInstanceAttribute.MultiTenancySides == MultiTenancySides.Tenant)
            {
                var contextMessage = filterContext.GetJobParameter<ContextMessage>("_ld_contextMessage");
                fingerprintKey = $"{contextMessage.TenantId}:{fingerprintKey}";
            }
            RemoveFingerprint(filterContext.Connection, fingerprintKey);
        }

        private static bool AddFingerprintIfNotExists(IStorageConnection connection, string fingerprintKey)
        {
            using (connection.AcquireDistributedLock(GetFingerprintLockKey(fingerprintKey), LockTimeout))
            {
                var fingerprint = connection.GetAllEntriesFromHash(GetFingerprintKey(fingerprintKey));

                if (fingerprint != null &&
                    fingerprint.ContainsKey("Timestamp") &&
                    DateTimeOffset.TryParse(fingerprint["Timestamp"], null, DateTimeStyles.RoundtripKind, out var timestamp) &&
                    DateTimeOffset.UtcNow <= timestamp.Add(FingerprintTimeout))
                {
                    // 有任務還未執行完，並且沒有超時
                    return false;
                }

                // 沒有任務執行，或者該任務已超時
                connection.SetRangeInHash(GetFingerprintKey(fingerprintKey), new Dictionary<string, string>
            {
                { "Timestamp", DateTimeOffset.UtcNow.ToString("o") }
            });

                return true;
            }
        }

        private static void RemoveFingerprint(IStorageConnection connection, string recurringJobId)
        {
            using (connection.AcquireDistributedLock(GetFingerprintLockKey(recurringJobId), LockTimeout))
            using (var transaction = connection.CreateWriteTransaction())
            {
                transaction.RemoveHash(GetFingerprintKey(recurringJobId));
                transaction.Commit();
            }
        }

        private static string GetFingerprintLockKey(string key)
        {
            return String.Format("{0}:lock", key);
        }

        private static string GetFingerprintKey(string key)
        {
            return String.Format("fingerprint:{0}", key);
        }


        void IClientFilter.OnCreated(CreatedContext filterContext)
        {
            
        }

        void IServerFilter.OnPerforming(PerformingContext filterContext)
        {
        }

    }