基于Quqrtz.NET 做的任務調度管理工具

作者：xling 2015-02-27 09:39:25

國慶前，需求讓我看了一下任務調度的數據表設計。和之前一樣，有100多個字段，p1 ~ p100, 我說這是干嘛啊！按這寫，寫死去了！然后在網上搜了一下開源的任務調度，第一個中意的就是 Quartz.NET，10.1 出來和老領導聚會，老領導說了另外一個東西：Zookeeper, 剛搜了一下，也有.NET的版本。

先入為主，Zookeeper 我就不深入了，整個10.1 在家基本除了看電影就是看 Quartz 了。

Quartz.net 提供了 Remoting 方式，Remoting 我08年的時候，寫過個小程序，讓一臺服務器通知另外一臺服務器去執行一些任務，簡單的應用，除了這些，在沒有深入過了。

看一下結構：

Jobs 下是任務

Listener 沒有寫好，不顯擺了。

QM.Server 是 Quartz 的調度服務

QM.Shell 是一個管理工具

我試想把管理工具放到本地來，即服務在服務器上，管理工具在本地，這樣就方便管理，但是除非本地能找到調度服務所用的所有的DLL，否則就會因為無法加載類型XXX而無法進行下去。

有了這個問題，那么只好管理工具和調度服務放一起了。

即然要把管理工具和服務放一起，那就不必把DLL COPY的到處都是，但是需要做一些特殊處理

下面是生成的目錄結構：

XXXXXX 管理工具目錄，

XXXXXX\Service 調度服務程序集存放在這里

XXXXXX\Service\Jobs 任務程序集存放在這里

要讓管理和調度服務都能找得到任務程序集，需要改一下 app.config

管理工具的 app.config

....

  <runtime>
    <assemblyBinding xmlns="urn:schemas-microsoft-com:asm.v1">
      <probing privatePath="Service/Jobs" />
...

調度服務的 app.config

...
  <runtime>
    <assemblyBinding xmlns="urn:schemas-microsoft-com:asm.v1">
      <probing privatePath="Jobs"/>
...

這樣一來，把新寫好的任務程序集放到 XXXXX\Service\Jobs 下，管理工具和調度服務都可以找得到了。

QM.Server 用到的東西：

Quartz.Net 2.2.4.400

log4net 1.2.10

Common.Logging

Common.Logging.log4net 由于 log4net 選用的是 1.2.10 ，所以這個只能用 2.0

Topshelf

列出來這些是因為我用 NuGet 搜索安裝的 Common.Logging.log4net ，程序一運行就退出。手動指定為 2.0 的 Common.Logging.log4net 才沒有問題。

調度服務我用的是 SqlCe 4.0 來存任務信息，所以需要安裝 SqlCe 4.0 的驅動, 可以從這里下載：

http://www.microsoft.com/en-us/download/details.aspx?id=17876

如果想指定其它數據庫，可以修改 quartz.config

quartz.jobStore.type = Quartz.Impl.AdoJobStore.JobStoreTX, Quartz 
quartz.jobStore.driverDelegateType = Quartz.Impl.AdoJobStore.StdAdoDelegate, Quartz 
quartz.jobStore.dataSource = ds 
quartz.dataSource.ds.connectionString = Data Source=QUARTZ.sdf;Persist Security Info=False; 
quartz.dataSource.ds.provider = SqlServerCe-400

因為管理工具是基于 Quartz REMOTING 的，所以要在 quartz.coonfig 中指定以下配置：

quartz.scheduler.exporter.type = Quartz.Simpl.RemotingSchedulerExporter, Quartz
quartz.scheduler.exporter.port = 5555
quartz.scheduler.exporter.bindName = TestScheduler
quartz.scheduler.exporter.channelType = tcp
quartz.scheduler.exporter.channelName = httpQuartz

這里我指定的端口號是 5555， bindName 是 TestScheduler, 這兩個東西要用在管理工具上。

Listener , Listener 不能通過 Remoting 來管理，因為 Not Support. 我嘗試做一個通UDP來廣播 Listener ，不知道哪里沒有寫好，就是收不到數據。沒有做好，我就不多廢話了。

管理工具用到的東西：

Quartz

Caliburn.Micro

看到 Caliburn 就知道這個管理工具是用 WPF 寫的了！沒什么講的，上圖看看：

連接界面，主機即調度服務所在的IP

端口和 Scheduler 是上面所講的配置中的 port 和 bindName

觸發器/任務列表，可以用來停止/啟動/編輯/刪除指定的任務。

日歷編輯界面，目前只實現了 HolidayCalendar 的編輯，Holiday 是用來指定任務在哪些日期不執行的。

這個功能 Cron 表達式無法做到，其它的基本都可以由 Cron 來做，所以目前只實現了 HolidayCalendar 的編輯。

這個是日歷列表界面，右邊會跟據不同的日歷類型顯示日歷詳細，只是目前只實現了 HolidayCalendar, 所以右邊只有這一個界面。

任務編輯界面，觸發器類型編輯器目前只實現了 Simple 和 Cron 類型的。

概覽。

源碼下載：

http://files.cnblogs.com/xling/QuartzJobManager.7z

要運行示例，需要運行 QM.Shell\bin\Debug\Service\QM.Server.exe 然后在運行 QM.Shell.exe 。

Quartz.NET 的API還是挺簡單的，還有許多細小功能沒有添加上。

另外還有一個分布式的，還沒有看相關資料，沒有處理。

#p#

界面具體變化如下:

任務參數可視化

如上圖所示, 在管理任務的界面上就可以知道這個任務需哪些參數/類型及參數的說明.

實現方式, 在 Job 上添加特性 : ParameterTypeAttribute

 1 namespace JobA {
 2     [ParameterType(typeof(Parameter))]
 3     public class Job : IJob {
 4 
 5         public static ILog Log = LogManager.GetLogger(typeof(Job));
 6         public void Execute(IJobExecutionContext context) {
 7             var dataMap = context.JobDetail.JobDataMap;
 8             //if (dataMap.ContainsKey("int")) {
 9             //    var pInt = dataMap.GetIntValue("int");
10             //    Console.WriteLine("1 JobA Parameter {0}", pInt);
11             //} else {
12             //    Log.Error("缺少參數 int, 未執行");
13             //    throw new JobExecutionException("缺少參數");
14             //}
15 
16             var p = dataMap.Parse<Parameter>();
17             Console.WriteLine("{0}\t{1}\t{2}\t{3}", p.PDateTime, p.PDecimal, p.PInt, p.PNullableInt);
18 
19 
20             Thread.Sleep(TimeSpan.FromMinutes(3));
21         }
22     }
23 }

取參數直接調用 dataMap.Parse<Parameter>() 就行了.

Parse 方法在: QM.Common. DatamapParser 中定義.

相比原始的從 DataMap 中用 key / value 方法取參數, 這種處理方式的好處不言而喻.

但是也有缺點， DataMap 支持任何可序列化的類型,

而用這種方法只支持

string, decimal, long, int, single, double, DateTime, DateTimeOffset, TimeSpan , bool, char 這些類型. (沒有做更深一步的處理, 有興趣的,可以嘗試自己去實現.)

每個任務獨立的應用程序域

試想一下插件式開發, 如果你做的插件需要N個第三方DLL, 而這些DLL并沒有引用到主項目上, 怎么辦呢? 一堆的 FileLoadException, FileNotFoundException 等錯誤, 想想都頭疼.

如果你開發的插件想擁有自己的配置文件, 又該怎么辦呢? 自己實現一個配置文件讀取解析? ini ? xml ? 頭疼吧.

針對上面的問題, 在這里的最佳解決辦法是 : 獨立的應用程序域.

這個要從 IScheduler.JobFactory 說起.

在QM.Server.QMServer 的構造方法中, 指定 Schedule.JobFactory 為 IsolatedJobFactory

IsolatedJobFactory 的定義:

 1     public class IsolatedJobFactory : IJobFactory {
 2 
 3         public IJob NewJob(TriggerFiredBundle bundle, IScheduler scheduler) {
 4             return NewJob(bundle.JobDetail.JobType);
 5         }
 6 
 7         private IJob NewJob(Type jobType) {
 8             return new IsolatedJob(jobType);
 9         }
10 
11         public void ReturnJob(IJob job) {
12             IDisposable disposable = job as IDisposable;
13             if (disposable != null) {
14                 disposable.Dispose();
15             }
16         }
17     }

從 NewJob 方法上可以看出, 實例出來的 Job 并不是最終要執行的 Job, 而是 IsolatedJob 的實例, 它類似中間人的身份.

IsolatedJob 實現了 IInterruptableJob 接口, 為中斷執行中的任務埋下伏筆.

在 IsolatedJob 的構造方法中, 通過 IsolateDomainLoader 新建一個應用程序域:

IsolatedDomainLoader 的構造函數:

 1 public IsolateDomainLoader(string path, string configFileName = "") {
 2     AppDomainSetup setup = new AppDomainSetup();
 3     setup.ApplicationName = "IsolateDomainLoader";
 4     setup.ApplicationBase = path;
 5     setup.DynamicBase = path;
 6     setup.PrivateBinPath = path;
 7     setup.CachePath = setup.ApplicationBase;
 8     setup.ShadowCopyFiles = "true";
 9     setup.ShadowCopyDirectories = setup.ApplicationBase;
10     if (!string.IsNullOrWhiteSpace(configFileName)) {
11         setup.ConfigurationFile = configFileName;
12         setup.ConfigurationFile = Path.Combine(path, configFileName);
13     }
14     this.Domain = AppDomain.CreateDomain("ApplicationLoaderDomain", null, setup);
15 }

參數 path 即最終要執行的 job 所在的 dll 的路徑.

configFileName 即獨立的配置文件名稱.

這樣一來, 一個 job 一個文件夾, 文件夾內放置這個 job 相關的DLL和配置文件, 和主程序完全隔離開來.

上面說 IsolatedJob 是個中間人, 這里解釋一下:

1, IsolatedJobFactory 的 NewJob 方法返回的是 IsolatedJob 的實例, 而不是最終要執行的 Job.

2, 在 IsolatedJob 中, 會通過獨立的應用程序域實例一個最終要執行的 Job 的遠程對象(通過 RemoteObject).

3, 當中間人的 Execute 方法被調用時, 會調用遠程 Job 對象的 Execute 方法.

4, Interrupt 方法同理.

遠程對象續約

因為獨立的應用程序域用到了遠程對象: MarshalByRefObject, 因此涉及到了遠程對象的租約過期及續租的問題.

遠程對象的租約默認為 5 分鐘, 可以重寫 InitializeLifetimeService 方法來修改租約的有效期. 但是一個 Job 不確定要執行多長時間, 修改租約有效期不是很合適, 所以這里是通過續約的方式來處理租約過期的問題.

本人對租約了解不多, 不多嘴.感興趣的話,可參見源碼:

QM.RemoteLoader.RemoteObjectSponsor 類

和 QM.RemoteLoader.IsolateDomainLoader類的 GetObject 方法.

立即中斷正在執行的任務

這個命題是有條件的, 即: 任務必須實現: IInterruptableJob 接口.

一般一個任務要執行很長時間, 如果不給個中斷的接口, 那就只能關閉服務或等任務執行完畢了.

實現了這個接口,在配合 CancellationToken.ThrowIfCancellationRequested 方法就可以中斷當前執行的任務了(別告訴我,你的任務是單線程的).

卸載域

任務執行完成后, 會將關聯的 IsolatedJob對象釋放, 在 IsolatedJob 的 Dispose 方法中,會把IsolateDomainLoader 對象釋放,IsolateDomainLoader 釋放的時候, 會把關聯的子應用程序域卸載.
所以, 如果如果你的任務是多線程的, 請在線程遠行完之前, 進行阻塞.

自定義Job的基類

目前, 如果自定義的 Job 的基類在第三方DLL中, 而且第三方DLL未引用到QM.Server 項目中, 并且不在 QM.Server\Jobs 目錄下, 會報:

未能加載文件或程序集 XXX 或它的某一個依賴項。系統找不到指定的文件。

解決辦法有兩種:

1, 將缺少的DLL放到Jobs 目錄下.

2, 將缺少的DLL添加引用到 QM.Server 中.

注意, 該限制只針對 Job 的基類. 除基類使用外的第三方DLL不需要這樣做, 在JOB上引用就是了.

放上一段不用的, 可終止的任務示例代碼給你做參考

[ParameterType(typeof(FetcherParameter))]  
    public class ScheduleFetcherJob : IInterruptableJob, IDisposable {  
 
 
        private CancellationTokenSource CTS = new CancellationTokenSource();  
 
        private long JobID = DateTime.Now.Ticks;  
        public void Execute(IJobExecutionContext context) {  
 
            var par = context.JobDetail.JobDataMap.Parse<FetcherParameter>();  
            this.CTS.Token.Register(() => {  
                Console.WriteLine("正在嘗試終止當前任務");  
            });  
            this.Execute(par);  
        }  
 
        private string GetUrl(string org, string dest) {  
            return string.Format("http://www.soushipping.com/shipping/{0}/{1}/{2}",  
                org, dest,  
                DateTime.Now.ToString("yyyy-MM-dd"));  
        }  
 
        private void Execute(FetcherParameter par) {  
            IFetcher<string> cityFetcher = new OrginCityFetcher();  
            var orgCities = cityFetcher.GetDatasByUrl(cityFetcher.InitUrl);  
            cityFetcher = new DestCityFetcher();  
            var destCities = cityFetcher.GetDatasByUrl(cityFetcher.InitUrl);  
 
            Console.WriteLine("找到 {0} 條始發地, {1} 條目的地", orgCities.Count(), destCities.Count());  
 
            var limitScd = new LimitedConcurrencyLevelTaskScheduler(par.MaxThread);  
 
            var factory = new TaskFactory(limitScd);  
 
            List<Task> tasks = new List<Task>();  
            foreach (var oc in orgCities.ToList()) {  
                foreach (var dc in destCities.ToList()) {  
                    //注意下面這句的參數 t, 如果帶這個參數,  IsCanceled 永遠都為 false  
                    //var task = Task.Factory.StartNew((t) => {  
                    var task = factory.StartNew(() => {  
                        this.CTS.Token.ThrowIfCancellationRequested();  
 
 
                        var url = this.GetUrl(oc, dc);  
 
                        var fetcher = new ScheduleFetcher(url);  
                        fetcher.PageFetchCompleted += fetcher_PageFetchCompleted;  
                        fetcher.DownloadCompleted += fetcher_DownloadCompleted;  
                        fetcher.Fetch();  
                        fetcher = null;  
                    }, this.CTS.Token)  
                    .ContinueWith(t => {  
                        //var completed = tasks.Where(tt => tt.Status == TaskStatus.RanToCompletion).Count();  
                        //Console.WriteLine("{0}\t已完成:{1}", DateTime.Now.ToString("yyyy/MM/dd"), completed);  
                        var arr = tasks.GroupBy(tt => tt.Status).Select(g => string.Format("{0}:{1}", g.Key, g.Count()));  
                        Console.WriteLine("{0}\t{1}", DateTime.Now.ToString("MM/dd HH:mm:ss"), string.Join("  ", arr));  
                        t.Dispose();  
                    });//  
                    //, TaskContinuationOptions.OnlyOnRanToCompletion)  
                    //.ContinueWith(t => {  
                    //    //Console.WriteLine("正在取消");  
                    //    t.Dispose();  
                    //}, TaskContinuationOptions.OnlyOnCanceled).ContinueWith(t => {  
                    //    Console.WriteLine("發生錯誤");  
                    //    t.Dispose();  
                    //}, TaskContinuationOptions.OnlyOnFaulted);  
 
                    tasks.Add(task);  
                }  
            }  
 
            try {  
                Task.WaitAll(tasks.ToArray());  
            } catch (AggregateException ex) {  
                ex.Handle(er => er is TaskCanceledException);  
            }  
            Console.WriteLine("任務完成");  
        }  
 
        #region  
        //private void Execute2(FetcherParameter par) {  
        //    IFetcher<string> cityFetcher = new OrginCityFetcher();  
        //    var orgCities = cityFetcher.GetDatasByUrl(cityFetcher.InitUrl);  
        //    cityFetcher = new DestCityFetcher();  
        //    var destCities = cityFetcher.GetDatasByUrl(cityFetcher.InitUrl);  
 
        //    Console.WriteLine("找到 {0} 條始發地, {1} 條目的地", orgCities.Count(), destCities.Count());  
 
        //    var urls = orgCities.SelectMany(o => destCities.Select(d => this.GetUrl(o, d)));  
 
        //    var opts = new ParallelOptions() {  
        //        MaxDegreeOfParallelism = par.MaxThread  
        //    };  
 
        //    var total = urls.Count();  
        //    object lockObj = new object();  
 
        //    //int sum = 0;  
        //    Parallel.ForEach(urls, opts,  
        //        (url) => {  
        //            var fetcher = new ScheduleFetcher(url);  
        //            fetcher.PageFetchCompleted += fetcher_PageFetchCompleted;  
        //            fetcher.DownloadCompleted += fetcher_DownloadCompleted;  
        //            fetcher.Fetch();  
        //            fetcher = null;  
 
        //            lock (lockObj) {  
        //                total--;  
        //                Console.WriteLine(total);  
        //            }  
        //        }  
        //        );  
 
        //}  
        #endregion  
 
        private void fetcher_DownloadCompleted(object sender, DownloadArgs e) {  
            if (e.ExceptionStatus.HasValue) {  
                Console.WriteLine("{0}\t請求地址: {1} 時,發生異常 {2}, 請檢查網絡環境.", DateTime.Now.ToString("yyyy/MM/dd HH:mm:ss"), e.Url, e.ExceptionStatus);  
            }  
        }  
 
 
        private List<DIRTY_SCHEDULE> Datas = new List<DIRTY_SCHEDULE>();  
        private object lockObj = new object();  
 
        void fetcher_PageFetchCompleted(object sender, FetchArgs<DIRTY_SCHEDULE> e) {  
            var datas = e.Datas.Distinct(d => d.UNQTAG);  
 
            lock (lockObj) {  
                this.Datas.AddRange(datas);  
                if (this.Datas.Count > 100) {  
                    var tmp = new DIRTY_SCHEDULE[this.Datas.Count];  
                    this.Datas.CopyTo(tmp);  
                    this.Datas = new List<DIRTY_SCHEDULE>();  
                    //不是放入線程池, 而是立即執行的線程  
                    var tr = new Thread(new ParameterizedThreadStart(this.SaveDatas));  
                    tr.Start(tmp);  
                }  
            }  
        }  
 
        private void SaveDatas(object state) {  
            IEnumerable<DIRTY_SCHEDULE> datas = (IEnumerable<DIRTY_SCHEDULE>)state;  
            var biz = new Biz.DirtyScheduleBiz();  
            biz.SaveDirtySchedule(datas, this.JobID);  
        }  
 
        public void Interrupt() {  
            this.CTS.Cancel();  
        }  
 
        ~ScheduleFetcherJob() {  
            Dispose(false);  
        }  
 
        public void Dispose() {  
            this.Dispose(true);  
            GC.SuppressFinalize(this);  
        }  
 
        protected virtual void Dispose(bool disposing) {  
            if (disposing) {  
                if (this.CTS != null)  
                    this.CTS.Dispose();  
 
                Console.WriteLine("Job Disposed");  
            }  
        }  
    }

最后, 源碼下載

https://github.com/gruan01/QM

謝謝圍觀, 新年快樂!

----------------------------

題外: 大年初二, 我手一抖, 把斷斷續續寫了快一年的東西給誤刪了!誤刪了啊!

用 360 的數據恢復功能, 沒錯, 是 360, 找出的文件, 我哭了, 數據庫(SQLCE) 恢復出來的文件損壞, 用SQLCE的修復工具修復, 是個空庫! 也就是說, 恢復出來的文件就是個屁!跟本就沒有恢復出來!

EXCEL 文件也一樣, 打不開!

更糟糕的是, 我沒有驗證, 恢復之后就直接蓋到原來的位置上了!

淚奔啊, 大過年的, 我就忙著干這個去了!

責任編輯：林師授來源： xling的博客

.NET Quqrtz.NET