asp.net core microservices 架构之分布式自动计算(二)

一简介

上一篇介绍了zookeeper如何进行分布式协调，这次主要讲解quartz使用zookeeper进行分布式计算，因为上一篇只是讲解原理，而这次实际使用，zookeeperService更改了一部分代码，算是集成优化吧。

系统结构图如下：

上图展示了，两个部分，一个是故障灾难转移集群，一个实现了分片的功能。故障灾难转移集群是quartz自带的功能，就不多说，分片功能是自己实现的。在这里要说下quartz使用故障灾难转移集群的一些配置注意事项：

再者就是netcore不支持remote，这个很重要，所以需要自己做一个web服务接口，但是本实例没有实现，而是仅仅使用数据库去配置和跟踪quartz服务，但是这是主要的。而使用api的一些功能就是实时开启，关闭，监控quartz主机状态，监控分片主机状态。所以大家留意这些功能暂时没有，不过大家在本文学会后很容易就可以自己扩展。

在这里要感谢 github账号为 weizhong1988/Weiz.TaskManager 的一个quartz管理的项目。

当然我这次的案例都是基于linux和mysql，而这个项目是sql server，所以我把sql全部替换了，更改了一些东西。后面会把代码全部放出来。界面如下图

好，下面看代码实现。

二 quartz故障灾难转移和分片功能

首先看结构：

然后看Program入口方法：

 var host = new HostBuilder()

                    .UseEnvironment(EnvironmentName.Development)

                    .ConfigureAppConfiguration((hostContext, configApp) =>

                    {

                        configApp.SetBasePath(Directory.GetCurrentDirectory());

                        configApp.AddJsonFile(

                              $"appsettings.{hostContext.HostingEnvironment.EnvironmentName}.json",

                                 optional: true);

                        configApp.AddEnvironmentVariables("PREFIX_");

                        configApp.AddCommandLine(args);

                        var rootConfiguration = configApp.Build();

                        QuartzOpt = new QuartzOption();

                        rootConfiguration.GetSection("Quartz").Bind(QuartzOpt); //绑定quartz的配置类的数据

                    }).ConfigureLogging((hostContext, configBuild) =>

                    {

                        configBuild.AddConfiguration(hostContext.Configuration.GetSection("Logging"));

                        configBuild.AddConsole();

                        configBuild.AddCustomizationLogger();

                    })

                    .ConfigureServices((hostContext, service) =>

                    {

                        service.AddKafka(KafkaBuilder =>

                        {

                            KafkaBuilder.AddConfiguration(hostContext.Configuration.GetSection("KafkaService"));

                        });

                        service.AddZookeeper(zookeeperBuilder =>

                        {

                            zookeeperBuilder.AddConfiguration(hostContext.Configuration.GetSection("zookeeperService"));

                        });

                        service.AddDbContext<QuartzDbContext>(option =>

                        option.UseMySQL(hostContext.Configuration.GetConnectionString("QuartzDatabase")), ServiceLifetime.Transient, ServiceLifetime.Transient);
　　　　　　　　　　　　　　//这个是操作数据库的数据库服务，这个和 quartz的cluster数据提供程序是分开的。

                    })

                    .Build();

            Host = host;

            ILoggerFactory loggerFact = host.Services.GetService<ILoggerFactory>();

            LogProvider.SetCurrentLogProvider(new ConsoleLogProvider(loggerFact)); //将框架的日志提供程序，传递给quart的日志接口。　　

　　　　　　　var ischema = RunProgramRunExample(loggerFact); //从数据库构造job的方法

            host.WaitForShutdown(); //netcore的通用主机。

            ischema.Shutdown(true);//quartz自己的主机。

quartz框架的架构和netcore目前的架构不兼容，netcore的通道和服务部件的软件架构方式，quartz先天不支持，你无法将任何上下文，比如host上下文，configuration上下文或者service上下文，传递给quartz。所以我使用了属性的方式：

 　　　　private ILoggerFactory _loggerFact;

        public static IHost Host { get; set; }

        public static String QUARTZ_INSTANCE_ID = "PREFIX_QUARTZ_INSTANCE_ID";

        public static QuartzOption QuartzOpt { get; set; }

在quartz上下文是这样使用的：

      ILoggerFactory loggerFact = Program.Host.Services.GetService<ILoggerFactory>();

            var  _logger=loggerFact.CreateLogger<ZookeeperService>();

而我在上一节说的不支持remote的解决方案，使用netcore的解决方案就是kestrel，netcoe宣称后面将要将webHost合并到通用主机里面，netcore确实目前发展较快，内部代码优化时，代码变动较大

，比如我上次扩展log模块，上一版本程序设计上和当前版本几乎没有什么可重用的，当然，对于开发者，并没有什么区别，因为肯定会保持兼容。

咱们看quart的日志模块：ConsoleLogProvider

using System;

using Microsoft.Extensions.Logging;

using Quartz.Logging;

namespace  Walt.Framework.Quartz.Host

 {

     public class ConsoleLogProvider : ILogProvider

    {

            private ILoggerFactory _logFactory;

            public ConsoleLogProvider(ILoggerFactory logFactory)

            {

                _logFactory=logFactory;

            }

            public Logger GetLogger(string name)

            {

                return (level, func, exception, parameters) =>

                {

                    if (func != null)

                    {

                        string logInfo=string.Format(func(), parameters);

                        var log=_logFactory.CreateLogger<ConsoleLogProvider>(); //将提供程序，替换为自定义的分布式log提供程序

                        log.LogDebug(logInfo);

                    }

                    return true;

                };

            }

            public IDisposable OpenNestedContext(string message)

            {

                throw new NotImplementedException();

            }

            public IDisposable OpenMappedContext(string key, string value)

            {

                throw new NotImplementedException();

            }

    }

 }

再看quartz的配置类：

namespace Walt.Framework.Quartz.Host

{

    public class QuartzOption

    {

        public string InsatanceId{get;set;}          //很重要，cluster中必须两个实例不一样。

        public string InstanceName{get;set;}          //quartz的实例名称，一般情况下用于显示名称。

        public bool IsClear{get;set;}                  //是否启动的时候清理job，因为cluster在数据库中有历史数据。

        public bool IsSlave{get;set;}                  //是不是slave，预留，暂时没用

        public int CustomerRecordCountForTest{get;set;} //分片时候，每个机器分到的需要处理的数据的数量

    }

}

然后就是咱们的主要方法：RunProgramRunExample

  private static IScheduler RunProgramRunExample(ILoggerFactory loggerFact)

        {

            var log = loggerFact.CreateLogger<Program>();

            try

            {

                var config = Host.Services.GetService<IConfiguration>();

                // Grab the Scheduler instance from the Factory

                NameValueCollection properties = new NameValueCollection

                {

                    ["quartz.scheduler.instanceName"] = QuartzOpt.InstanceName,

                    ["quartz.scheduler.instanceId"] = QuartzOpt.InsatanceId,

                    ["quartz.threadPool.type"] = "Quartz.Simpl.SimpleThreadPool, Quartz",

                    ["quartz.threadPool.threadCount"] = "",

                    ["quartz.jobStore.misfireThreshold"] = "",

                    ["quartz.jobStore.type"] = "Quartz.Impl.AdoJobStore.JobStoreTX, Quartz",

                    ["quartz.jobStore.useProperties"] = "false",

                    ["quartz.jobStore.dataSource"] = "default",

                    ["quartz.jobStore.tablePrefix"] = "QRTZ_",

                    ["quartz.jobStore.clustered"] = "true",

                    ["quartz.jobStore.driverDelegateType"] = "Quartz.Impl.AdoJobStore.MySQLDelegate, Quartz",

                    ["quartz.dataSource.default.connectionString"] = config.GetConnectionString("QuatrzClustDatabase"),

                    ["quartz.dataSource.default.provider"] = "MySql",

                    ["quartz.serializer.type"] = "json",

                    ["quartz.scheduler.exporter.type"] = "Quartz.Simpl.RemotingSchedulerExporter, Quartz",  
　　　　　　　　　    //从这个往下netcore都不支持，以上为集群配置，只要两个实例id不同的quartz，配置同一个数据源
                   ，就会自动的按照cluster运行，还有一点就是如果分布到不同机器，一定要配置ntp时间服务器，同步时间。

                    ["quartz.scheduler.exporter.port"] = "",

                    ["quartz.scheduler.exporter.bindName"] = "QuartzScheduler",

                    ["quartz.scheduler.exporter.channelType"] = "tcp",

                    ["quartz.scheduler.exporter.channelName"] = "httpQuartz",

                    ["quartz.scheduler.exporter.rejectRemoteRequests"] = "true"

                };

                StdSchedulerFactory factory = new StdSchedulerFactory(properties);

                IScheduler scheduler = factory.GetScheduler().GetAwaiter().GetResult();

                 string machine = Environment.MachineName;  //获取当前的机器名

                 QuartzDbContext db = Host.Services.GetService<QuartzDbContext>();

                var listQuartzTask = db.QuartzTask.Where(w => w.IsDelete == 0

                && w.MachineName == machine && w.InstanceId == QuartzOpt.InsatanceId)

                .ToListAsync().GetAwaiter().GetResult();  //从数据库中获取这台机器和实例中的job

                log.LogDebug("从数据库获取task记录,详细信息:{0}", Newtonsoft.Json.JsonConvert.SerializeObject(listQuartzTask));

                Dictionary<string,Assembly> collAssembly=new Dictionary<string, Assembly>();  //加载程序集

                foreach (var item in listQuartzTask)//首先第一次加载全部的程序集

                {

                    //加载程序集

                    if (!string.IsNullOrEmpty(item.AssemblyName) && !collAssembly.ContainsKey(item.AssemblyName))

                    {

                        try

                        {

                            collAssembly[item.AssemblyName] =

                            AssemblyHelp.GetAssemblyByteByAssemblyName(

                                Path.Combine(Directory.GetCurrentDirectory(), "AssemblyColl"), item.AssemblyName);

                        }

                        catch (Exception ep)

                        {

                            log.Log(Microsoft.Extensions.Logging.LogLevel.Error, , ep, "没有找到程序集.");

                            Task.Delay();

                            continue;

                        }

                    }

                }

                // and start it off

                scheduler.Start();

                // if (!QuartzOpt.IsSlave)

                // {

                    var task = Task.Run(() =>

                    {

                        bool isClear = QuartzOpt.IsClear;

                        log.LogInformation("job监控程序开始循环，间隔为15秒");

                        while (true) //主要用来循环数据库记录，在添加或者修改job的时候，自动重新添加和执行job。

                        {

                            try

                            {

                                if (scheduler != null)

                                {

                                    log.LogDebug("检查scheduler是否开始");

                                    if (scheduler.IsStarted)

                                    {

                                        if (isClear) //启动清理

                                        {

                                            scheduler.Clear().GetAwaiter().GetResult();

                                            isClear = false;

                                        }

                                        log.LogDebug("scheduler已经开始");

                                        db = Host.Services.GetService<QuartzDbContext>();

                                        listQuartzTask = db.QuartzTask.Where(w => w.IsDelete == 0

                                            && w.MachineName == machine && w.InstanceId == QuartzOpt.InsatanceId)

                                            .ToListAsync().GetAwaiter().GetResult(); //在循环中获取数据库中本机器和实例的job记录。

                                        log.LogDebug("从数据库获取task记录,详细信息:{0}", Newtonsoft.Json.JsonConvert.SerializeObject(listQuartzTask));

foreach (var item in listQuartzTask)

                                        {//加载程序集

                                            if (!string.IsNullOrEmpty(item.AssemblyName) && !collAssembly.ContainsKey(item.AssemblyName)) //预先加载新添加的job的程序集

                                            {

                                                try

                                                {

                                                    collAssembly[item.AssemblyName] =

                                                    AssemblyHelp.GetAssemblyByteByAssemblyName(

                                                        Path.Combine(Directory.GetCurrentDirectory(), "AssemblyColl"), item.AssemblyName);

                                                }

                                                catch (Exception ep)

                                                {

                                                    log.Log(Microsoft.Extensions.Logging.LogLevel.Error, , ep, "没有找到程序集.");

                                                    Task.Delay();

                                                    continue;

                                                }

                                            }

                                            log.LogDebug("开始检查task：{0}", Newtonsoft.Json.JsonConvert.SerializeObject(item));

                                            var jobKey = new JobKey(item.TaskName, item.GroupName);

                                            var triggerKey = new TriggerKey(item.TaskName, item.GroupName);

                                            if (scheduler.CheckExists(jobKey).Result) //如果存在，则根据状态处理相应的动作。

                                            {

                                                var jobDetai = scheduler.GetJobDetail(jobKey);

                                                var trigger = scheduler.GetTrigger(triggerKey);

                                                log.LogDebug("此task已经存在scheduler中，数据库状态：{0}，scheduer中的状态：{1}.trigger状态：{2}"

                                                , ((OperateStatus)item.OperateStatus).ToString(), jobDetai.Status.ToString(), trigger.Status.ToString());

                                                if ((OperateStatus)item.OperateStatus == OperateStatus.Stop) //如果数据库中停止job，则删除这个job，如果有remote，可以实时，处理，这断代码就没有用了，但是可以作为远程处理失败的预防错误，所以可以保留。

                                                {

                                                    log.LogInformation("删除schduler中的job：{0}", jobKey.ToString());

                                                    if (!scheduler.DeleteJob(jobKey).GetAwaiter().GetResult())

                                                    {

                                                        log.LogError("删除job失败。name:{0},group:{1}", jobKey.Name, jobKey.Group);

                                                    }

                                                }

                                                else

                                                {

                                                    if (jobDetai.IsFaulted) //如果失败，则更改数据库中job的状体，同理，如果有remote，这个可以作为预防错误。

                                                    {

                                                        if (jobDetai.Exception != null)

                                                        {

                                                            log.LogError(, jobDetai.Exception, "job faulted");

                                                        }

                                                        var jobItem = db.QuartzTask.FirstOrDefault(w => w.IsDelete ==

                                                        && w.TaskName == jobKey.Name

                                                        && w.GroupName == jobKey.Group

                                                        && w.MachineName == machine

                                                        && w.InstanceId == scheduler.SchedulerInstanceId);

                                                        item.Status = (int)TaskStatus.Faulted;

                                                        item.OperateStatus = (int)OperateStatus.Stop;

                                                        db.Update<QuartzTask>(jobItem);

                                                        db.SaveChanges();

                                                    }

                                                    else //如果非执行状态，则中断

                                                    {

                                                        if (jobDetai.Status != TaskStatus.Running

                                                            && jobDetai.Status != TaskStatus.RanToCompletion

                                                            && jobDetai.Status != TaskStatus.WaitingForActivation

                                                            && jobDetai.Status != TaskStatus.WaitingForChildrenToComplete

                                                            && jobDetai.Status != TaskStatus.WaitingToRun)

                                                        {

                                                            var interTask = scheduler.Interrupt(jobKey, new CancellationToken(true))

                                                            .GetAwaiter().GetResult();

                                                            jobDetai.Start();

                                                        }

                                                    }

                                                }

                                                var triggerListener = scheduler.ListenerManager.GetTriggerListener("triggerUpdate");

                                                if (triggerListener == null)

                                                {

                                                    triggerListener = new TriggerUpdateListens("trigger"+item.TaskName);

                                                    IMatcher<TriggerKey> triggermatcher = KeyMatcher<TriggerKey>.KeyEquals(triggerKey);

                                                    scheduler.ListenerManager.AddTriggerListener(triggerListener, triggermatcher);

                                                }

                                                var jobListener = scheduler.ListenerManager.GetJobListener("jobupdateListens");

                                                if (jobListener == null)

                                                {

                                                    IJobListener jobUpdateListener = new JobUpdateListens("job"+item.TaskName);

                                                    IMatcher<JobKey> jobmatcher = KeyMatcher<JobKey>.KeyEquals(jobKey);

                                                    scheduler.ListenerManager.AddJobListener(jobUpdateListener, jobmatcher);

                                                }

                                            }

                                            else //如果不存在，则新添加job，以及执行

                                            {

                                                log.LogInformation("添加新的job，判断是否状态为停止。");

                                                if ((OperateStatus)item.OperateStatus != OperateStatus.Stop)

                                                {

                                                    log.LogInformation("添加新的job");

                                                    var assemblyName = item.AssemblyName;

                                                    var className = item.ClassName;

                                                    Type jobTaskType = null;

                                                    try

                                                    {

                                                        jobTaskType = AssemblyHelp.GetTypeByAssemblyNameAndClassName(collAssembly[item.AssemblyName], className);

                                                        log.LogInformation("找到类型，type:{0}",className);

                                                    }

                                                    catch (Exception ep)

                                                    {

                                                        log.Log(Microsoft.Extensions.Logging.LogLevel.Error, , ep, "没有找到type.");

                                                    }

                                                    if (jobTaskType == null)

                                                    {

                                                        try

                                                        {

                                                            jobTaskType = AssemblyHelp

                                                            .GetTypeByCurrentAssemblyNameAndClassName(className, Assembly.GetExecutingAssembly());

                                                            if (jobTaskType == null)

                                                            {

                                                                log.LogInformation("没有找到类型");

                                                                continue;

                                                            }

                                                            log.LogInformation("找到类型，type:{0}",className);

                                                        }

                                                        catch (Exception ep)

                                                        {

                                                            log.Log(Microsoft.Extensions.Logging.LogLevel.Error, , ep, "没有找到类型.");

                                                            continue;

                                                        }

                                                    }

                                                    IJobDetail job = JobBuilder.Create(jobTaskType)

                                                        .WithIdentity(item.TaskName, item.GroupName)

                                                        .Build();

                                                    ITrigger trigger = TriggerBuilder.Create()

                                                        .WithIdentity(item.TaskName, item.GroupName)

                                                        .StartNow()

                                                        .WithCronSchedule(item.CronExpressionString)

                                                        .Build();

                                                    scheduler.ScheduleJob(job, trigger).GetAwaiter().GetResult();

                                                    log.LogInformation("添加成功，type:{0}",className);

                                                    ITriggerListener triggerListener = new TriggerUpdateListens("trigger"+item.TaskName);

                                                    IMatcher<TriggerKey> triggermatcher = KeyMatcher<TriggerKey>.KeyEquals(trigger.Key);

                                                    scheduler.ListenerManager.AddTriggerListener(triggerListener, triggermatcher);

                                                    IJobListener jobUpdateListener = new JobUpdateListens("job"+item.TaskName);

                                                    IMatcher<JobKey> jobmatcher = KeyMatcher<JobKey>.KeyEquals(job.Key);

                                                    scheduler.ListenerManager.AddJobListener(jobUpdateListener, jobmatcher);

                                                }

                                            }

                                        }

                                    }

                                    else

                                    {

                                        log.LogInformation("scheduler is not IsStarted");

                                    }

                                }

                                else

                                {

                                    log.LogInformation("scheduler is null");

                                }

                            }

                            catch (Exception ep)

                            {

                                log.Log(Microsoft.Extensions.Logging.LogLevel.Error, , ep, "task监控程序执行错误.");

                            }

                            Thread.Sleep();

                        }

                    });

                // }

                // else

                // {

                //     db = Host.Services.GetService<QuartzDbContext>();

                //     listQuartzTask = db.QuartzTask.Where(w => w.IsDelete == 0

                //                                             && w.MachineName == machine

                //                                             && w.InstanceId == QuartzOpt.InsatanceId)

                //                                       .ToListAsync().GetAwaiter().GetResult();

                //     foreach (var item in listQuartzTask)

                //     {

                //          var jobKey = new JobKey(item.TaskName, item.GroupName);

                //          var triggerKey = new TriggerKey(item.TaskName, item.GroupName);

                //         // var jobItem = db.QuartzTask.FirstOrDefault(w => w.IsDelete == 0

                //         //                          && w.TaskName == jobKey.Name

                //         //                          && w.GroupName == jobKey.Group

                //         //                          && w.MachineName == machine

                //         //                          && w.InstanceId == scheduler.SchedulerInstanceId);

                //         // item.Status = (int)TaskStatus.Faulted;

                //         // item.OperateStatus = (int)OperateStatus.Stop;

                //         // db.Update<QuartzTask>(jobItem);

                //         // db.SaveChanges();

                //         if (scheduler.CheckExists(jobKey).Result)

                //         {

                //             var triggerListener = scheduler.ListenerManager.GetTriggerListener("triggerUpdate");

                //             if (triggerListener == null)

                //             {

                //                 triggerListener = new TriggerUpdateListens();

                //                 IMatcher<TriggerKey> triggermatcher = KeyMatcher<TriggerKey>.KeyEquals(triggerKey);

                //                 scheduler.ListenerManager.AddTriggerListener(triggerListener, triggermatcher);

                //             }

                //             var jobListener = scheduler.ListenerManager.GetJobListener("jobupdateListens");

                //             if (jobListener == null)

                //             {

                //                 IJobListener jobUpdateListener = new JobUpdateListens();

                //                 IMatcher<JobKey> jobmatcher = KeyMatcher<JobKey>.KeyEquals(jobKey);

                //                 scheduler.ListenerManager.AddJobListener(jobUpdateListener, jobmatcher);

                //             }

                //         }

                //     }

                //}

                return scheduler;

                // Tell quartz to schedule the job using our trigger

                //await scheduler.ScheduleJob(job, trigger);

            }

            catch (SchedulerException sep)

            {

                log.Log(Microsoft.Extensions.Logging.LogLevel.Error, , sep, "job执行错误。");

            }

            return null;

        }

咱们现在看trriger监控类和job的监控类：

using System;

using System.Threading;

using System.Threading.Tasks;

using Quartz;

using Quartz.Logging;

using Quartz.Impl;

using Walt.Framework.Service.Zookeeper;

using Microsoft.Extensions.DependencyInjection;

using org.apache.zookeeper;

using System.Linq;

using System.Collections.Generic;

using Microsoft.Extensions.Logging;

using static org.apache.zookeeper.KeeperException;

namespace Walt.Framework.Quartz.Host

{

    public class TriggerUpdateListens : ITriggerListener

    {

        public string Name { get; set; }

        public TriggerUpdateListens(string name)

        {

            Name = name;

        }

        private bool VoteJob{ get; set;}

        public Task TriggerComplete(ITrigger trigger, IJobExecutionContext context, SchedulerInstruction triggerInstructionCode, CancellationToken cancellationToken = default(CancellationToken))

        {

            ILoggerFactory loggerFact = Program.Host.Services.GetService<ILoggerFactory>();

            var  _logger=loggerFact.CreateLogger<ZookeeperService>();

             _logger.LogInformation(, null, "执行成功.name:{0} group：{1}", context.JobDetail.Key.Name, context.JobDetail.Key.Group);

            return Task.FromResult(true);

        }

　　　　　//trigger激发，这是job执行的第一个执行的。

        public Task TriggerFired(ITrigger trigger, IJobExecutionContext context, CancellationToken cancellationToken = default(CancellationToken))

        {

            ILoggerFactory loggerFact = Program.Host.Services.GetService<ILoggerFactory>();

            var  _logger=loggerFact.CreateLogger<ZookeeperService>();

            _logger.LogInformation(, null, "开始执行job.name:{0} group：{1}", context.JobDetail.Key.Name, context.JobDetail.Key.Group);

            string machine = Environment.MachineName;  //获取当前机器名

            try

            {

                var customerAttri = context.JobDetail.JobType.GetCustomAttributes(false);

                foreach (var customer in customerAttri)

                {

                    if (customer is DistributingAttributes) //如果打这个标签，则说明是分片job。

                    {

                        var distri = customer as DistributingAttributes;

                        var zookeeper = Program.Host.Services.GetService<IZookeeperService>();

                        string currentTempNodeName = string.Empty;

                        string fullPath = "/lock/"+ context.JobDetail.Key.Name + context.JobDetail.Key.Group;

                        int flag = ;

                    Repeat: //这里因为某些原因失败，可以给重复几次。

                        string jsonData = zookeeper.GetDataByLockNode(fullPath, "getlock"

                        , ZooDefs.Ids.OPEN_ACL_UNSAFE, out currentTempNodeName);

                        if(jsonData==null)

                        {

                            _logger.LogError("获取锁失败。节点：{0},锁前缀：{1},重试：{2}",fullPath,"getlock",flag);

                            if(flag<=)

                            {

                                flag = flag + ;

                                goto Repeat;

                            }

                            VoteJob = true; //如果获取失败，则否决执行job，这个变量在下面的trriger方法中使用。

                            //context.Scheduler.Interrupt(context.JobDetail.Key);

　　　　　　　　　　　　　　　　　　return Task.FromResult(false); //返回false，则会执行VetoJobExecution方法。

                       }

　　　　　　　　　　　　　　//获取锁成功，处理分片数据，构造分片上下文。

                        QuartzDbContext db = Program.Host.Services.GetService<QuartzDbContext>();

                         var item = db.QuartzTask.Where(w => w.IsDelete ==

                        && w.TaskName == context.JobDetail.Key.Name

                        && w.GroupName == context.JobDetail.Key.Group

                        && w.MachineName == machine

                        && w.InstanceId == context.Scheduler.SchedulerInstanceId).FirstOrDefault();

                        if (item != null)

                        {

                            //TODO 这里可以找出机器名，拼接remote的api，可以查看分片主机是否存活，从而将一些挂起的任务重新分配。

                        }

                        string distributeFlag = item.MachineName + item.InstanceId;

                        List<DistributingData> distriData = new List<DistributingData>();

                        DistributingData currentDistriEntity = null;

                        if (string.IsNullOrEmpty(jsonData))

                        {

                            currentDistriEntity= new DistributingData  //分片元数据

                            {

                                DistributeFlag =distributeFlag, //分片标记，以机器名和实例名构造

                                PageIndex = ,

                                PageSize = Program.QuartzOpt.CustomerRecordCountForTest //配置的需要处理的数据数量

                            };

                            distriData.Add(currentDistriEntity);

                        }

                        else

                        {

                            distriData = Newtonsoft.Json.JsonConvert.DeserializeObject<List<DistributingData>>(jsonData);

                            if (distriData == null || distriData.Count() < )

                            {

                                currentDistriEntity= new DistributingData

                                {

                                    DistributeFlag =distributeFlag,

                                    PageIndex = ,

                                    PageSize = Program.QuartzOpt.CustomerRecordCountForTest //配置

                                };

                                distriData.Add(currentDistriEntity);

                            }

                            else

                            {

                                currentDistriEntity= distriData.Where(w => w.DistributeFlag == distributeFlag).SingleOrDefault();

                                if (currentDistriEntity == null) //当前主机还没有分片过，将当前主机加入分片集群

                                {

                                    var maxPageIndex = distriData.Max(w => w.PageIndex);

                                    maxPageIndex = maxPageIndex + ;

                                    var entity = new DistributingData

                                    {

                                        DistributeFlag = distributeFlag,

                                        PageIndex = maxPageIndex,

                                        PageSize = Program.QuartzOpt.CustomerRecordCountForTest //配置

                                    };

                                    distriData.Add(entity);

                                }

                                else

                                {

                                    var maxPageIndex = distriData.Max(w => w.PageIndex);

                                    maxPageIndex = maxPageIndex + ;

                                    currentDistriEntity.PageIndex = maxPageIndex;

                                }

                            }

                        }

                        item.Remark = Newtonsoft.Json.JsonConvert.SerializeObject(currentDistriEntity);

                        db.Update(item);

                        db.SaveChanges();

                        string resultData = Newtonsoft.Json.JsonConvert.SerializeObject(distriData);

                        context.JobDetail.JobDataMap.Put("distriData", currentDistriEntity); //将分片数据放入数据上下文，job中可以访问。

                        zookeeper.SetDataAsync(fullPath

                            , resultData, false).GetAwaiter().GetResult();

                        zookeeper.DeleteNode(currentTempNodeName); //处理完成，需要删除当前节点，释放锁。

                        _logger.LogInformation("分片执行：{0}",resultData);

                    }

                }

            }

             catch(ConnectionLossException cle)

            {

                VoteJob = true;

                _logger.LogError(cle, "获取同步锁出现错误。连接丢失");

            }

            catch(SessionExpiredException sep)

            {

                VoteJob = true;

                _logger.LogError(sep, "获取同步锁出现错误。连接过期");

            }

            catch(KeeperException kep)

            {

                VoteJob = true;

                _logger.LogError(kep, "获取同步锁出现错误。操作zookeeper出错");

            }

            catch(Exception ep)

            {

                try

                {

                     _logger.LogError(,ep,"分片失败。");

                    //context.Scheduler.DeleteJob(context.JobDetail.Key).GetAwaiter().GetResult();

                    VoteJob = true;

                     QuartzDbContext db = Program.Host.Services.GetService<QuartzDbContext>();

                    var item = db.QuartzTask.Where(w => w.IsDelete ==

                     && w.TaskName == context.JobDetail.Key.Name

                     && w.GroupName == context.JobDetail.Key.Group

                     && w.MachineName == machine

                     && w.InstanceId == context.Scheduler.SchedulerInstanceId).FirstOrDefault();

                    if (item == null)

                    {

                        _logger.LogError(, ep, "分片失败，获取数据库记录失败。");

                    }

                    else

                    {

                        item.Status = (int)TaskStatus.Canceled;

                        item.OperateStatus = (int)OperateStatus.Stop;

                        item.Remark = ep.ToString();

                        db.Update(item);

                        db.SaveChanges();

                    }

                }

                catch (Exception eep)

                {

                     _logger.LogError(, eep, "分片失败，更新数据库失败。");

                }

            }

            return Task.FromResult(true);

        }

        public Task TriggerMisfired(ITrigger trigger, CancellationToken cancellationToken = default(CancellationToken))

        {

            return Task.FromResult(true);

        }

　　　　　//

　　　　　//当TriggerComplete返回false，执行这个方法。

public Task<bool> VetoJobExecution(ITrigger trigger, IJobExecutionContext context, CancellationToken cancellationToken = default(CancellationToken))

        {

            ILoggerFactory loggerFact = Program.Host.Services.GetService<ILoggerFactory>();

            var  _logger=loggerFact.CreateLogger<ZookeeperService>();

            if (VoteJob)

            {

                _logger.LogInformation(, null, "取消执行job.name:{0} group：{1}", context.JobDetail.Key.Name, context.JobDetail.Key.Group);

            }

            return Task.FromResult(VoteJob);//VoteJob在TriggerComplete中指定，默认为false
　　　　　　　　　　　　　　　　　　　　　　　　　　//，如果获取锁失败，则设置为true，这个方法返回true，则只执行JobUpdateListens的JobExecutionVetoed方法，然后job这一次将不执行。

} } }

接下来看job执行前都执行那些方法：

using System;

using System.Threading;

using System.Threading.Tasks;

using Quartz;

using Quartz.Logging;

using Quartz.Impl;

using Microsoft.Extensions.DependencyInjection;

using System.Linq;

using Microsoft.EntityFrameworkCore;

using Microsoft.Extensions.Logging;

namespace Walt.Framework.Quartz.Host

{

    public class JobUpdateListens : IJobListener

    {

        public string Name { get; set; }

        public JobUpdateListens(string name)

        {

            Name = name;

        }

        public Task JobExecutionVetoed(IJobExecutionContext context, CancellationToken cancellationToken = default(CancellationToken))

        {

            return Task.FromResult(true);

        }

//job执行前执行，将状态放入数据库。

        public Task JobToBeExecuted(IJobExecutionContext context, CancellationToken cancellationToken = default(CancellationToken))

        {

            try

            {

                string machine = Environment.MachineName;

                QuartzDbContext db = Program.Host.Services.GetService<QuartzDbContext>();

                var item = db.QuartzTask.FirstOrDefault(w => w.IsDelete ==

                && w.TaskName == context.JobDetail.Key.Name

                && w.GroupName == context.JobDetail.Key.Group

                && w.MachineName == machine

                && w.InstanceId == context.Scheduler.SchedulerInstanceId);

                item.Status = (int)TaskStatus.WaitingToRun;

                db.Update<QuartzTask>(item);

                db.SaveChanges();

            }

            catch (Exception ep)

            {

                //context.Scheduler.Interrupt(context.JobDetail.Key);

                var logFaoctory = Program.Host.Services.GetService<ILoggerFactory>();

                var log = logFaoctory.CreateLogger<JobUpdateListens>();

                log.LogError(, ep, "JobToBeExecuted:Job执行错误,name：{0},Group:{1}", context.JobDetail.Key.Name, context.JobDetail.Key.Group);

            }

            return Task.FromResult(true);

        }

//job执行后执行，这个方法将执行结果放入数据库，处理异常。

        public Task JobWasExecuted(IJobExecutionContext context, JobExecutionException jobException, CancellationToken cancellationToken = default(CancellationToken))

        {

            try

            {

                QuartzDbContext db = Program.Host.Services.GetService<QuartzDbContext>();

                var logFaoctory = Program.Host.Services.GetService<ILoggerFactory>();

                var log = logFaoctory.CreateLogger<JobUpdateListens>();

                string machine = Environment.MachineName;

                var item = db.QuartzTask.FirstOrDefault(w => w.IsDelete ==

                                                        && w.TaskName == context.JobDetail.Key.Name

                                                        && w.GroupName == context.JobDetail.Key.Group

                                                        && w.MachineName == machine

                                                        && w.InstanceId == context.Scheduler.SchedulerInstanceId);

                if (jobException != null)

                {

                    item.Status = (int)TaskStatus.Faulted;

                    item.Remark = Newtonsoft.Json.JsonConvert.SerializeObject(jobException);

                    log.LogError("Job执行错误,name：{0},Group:{1}", context.JobDetail.Key.Name, context.JobDetail.Key.Group);

                }

                else

                {

                    item.Status = (int)TaskStatus.RanToCompletion;

                    item.RecentRunTime = context.FireTimeUtc.DateTime;

                    if (context.NextFireTimeUtc.HasValue)

                    {

                        item.NextFireTime = context.NextFireTimeUtc.Value.DateTime;

                    }

                }

                db.Update<QuartzTask>(item);

                db.SaveChanges();

            }

            catch (Exception ep)

            {

                //context.Scheduler.Interrupt(context.JobDetail.Key);

                var logFaoctory = Program.Host.Services.GetService<ILoggerFactory>();

                var log = logFaoctory.CreateLogger<JobUpdateListens>();

                log.LogError(, ep, "JobWasExecuted:Job执行错误,name：{0},Group:{1}", context.JobDetail.Key.Name, context.JobDetail.Key.Group);

            }

            return Task.FromResult(true);

        }

    }

}

在这次使用 zookeeperservice中，优化了一些代码：

using System;

using System.Collections.Generic;

using System.Threading.Tasks;

using Microsoft.Extensions.Logging;

using Microsoft.Extensions.Options;

using org.apache.zookeeper;

using org.apache.zookeeper.data;

using static org.apache.zookeeper.ZooKeeper;

using System.Linq;

using System.Linq.Expressions;

using System.Threading;

using static org.apache.zookeeper.Watcher.Event;

using Newtonsoft.Json;

using System.Collections.Concurrent;

using static org.apache.zookeeper.ZooDefs;

using static org.apache.zookeeper.KeeperException;

namespace  Walt.Framework.Service.Zookeeper

{

    internal class WaitLockWatch : Watcher

    {

        private AutoResetEvent _autoResetEvent;

        private ManualResetEvent _mutex;

        private ILogger _logger;

        private string _path;

        public WaitLockWatch(AutoResetEvent autoResetEvent

        , ILogger logger, string path

        , ManualResetEvent mutex)

        {

            _autoResetEvent = autoResetEvent;

            _logger = logger;

            _path = path;

            _mutex = mutex;

        }

        public override Task process(WatchedEvent @event)

        {

             _mutex.Set();

            return Task.FromResult(true);

        }

    }

    internal class WaitConnWatch : Watcher

    {

        private AutoResetEvent _autoResetEvent;

        private ILogger _logger;

        private ManualResetEvent _mutex;

        public WaitConnWatch(AutoResetEvent autoResetEvent

        ,ILogger logger

        ,ManualResetEvent mutex)

        {

            _autoResetEvent=autoResetEvent;

            _logger=logger;

            _mutex = mutex;

        }

       public override Task process(WatchedEvent @event)

       {

           _logger.LogInformation("watch激发,回掉状态：{0}",@event.getState().ToString());

            if(@event.getState()== KeeperState.SyncConnected

            ||@event.getState()== KeeperState.ConnectedReadOnly)

            {

                _logger.LogInformation("释放连接阻塞");

                _autoResetEvent.Set();

            }

            else

            {

                _logger.LogInformation("连接断开，释放分布式锁阻塞");

                _mutex.Set();

            }

            return Task.FromResult();

       }

    }

    public class ZookeeperService : IZookeeperService

    {

        private ZookeeperOptions _zookeeperOptions;

        private ZooKeeper _zookeeper;

         private static readonly byte[] NO_PASSWORD = new byte[];

         public Watcher Wathcer {get;set;}

         public ILoggerFactory LoggerFac { get; set; }

         private ILogger _logger;

        internal Thread CurrThread{ get; }

        AutoResetEvent[] autoResetEvent=new AutoResetEvent[]

         {new AutoResetEvent(false),new AutoResetEvent(false)};

        ManualResetEvent _manualReset = new ManualResetEvent(false);

        public ZookeeperService(IOptionsMonitor<ZookeeperOptions>  zookeeperOptions

        ,ILoggerFactory loggerFac)

        {

            LoggerFac=loggerFac;

            _logger=LoggerFac.CreateLogger<ZookeeperService>();

            _zookeeperOptions=zookeeperOptions.CurrentValue;

            _logger.LogInformation("配置参数：{0}",JsonConvert.SerializeObject(_zookeeperOptions));

             zookeeperOptions.OnChange((zookopt,s)=>{

                _zookeeperOptions=zookopt;

            });

            _logger.LogInformation("开始连接");

            Conn(_zookeeperOptions);

            CurrThread = System.Threading.Thread.CurrentThread;

        }

        private void Conn(ZookeeperOptions zookeeperOptions)

        {

            bool isReadOnly=default(Boolean);

            Wathcer=new WaitConnWatch(autoResetEvent[],_logger,_manualReset);

            if(isReadOnly!=zookeeperOptions.IsReadOnly)

            {

                isReadOnly=zookeeperOptions.IsReadOnly;

            }

            byte[] pwd=new byte[];

            //如果没有密码和sessionId

            if(string.IsNullOrEmpty(zookeeperOptions.SessionPwd)

            &&_zookeeperOptions.SessionId==default(int))

            {

             _zookeeper=new ZooKeeper(zookeeperOptions.Connectstring,zookeeperOptions.SessionTimeout,Wathcer,isReadOnly);

            }

            else if (!string.IsNullOrEmpty(zookeeperOptions.SessionPwd))

            {

                pwd=System.Text.Encoding.Default.GetBytes(zookeeperOptions.SessionPwd);

                 _zookeeper=new ZooKeeper(zookeeperOptions.Connectstring,zookeeperOptions.SessionTimeout,Wathcer,,pwd,isReadOnly);

            }

            else

            {

                 _zookeeper=new ZooKeeper(zookeeperOptions.Connectstring

                 ,zookeeperOptions.SessionTimeout,Wathcer,zookeeperOptions.SessionId,pwd,isReadOnly);

            }

             if(_zookeeper.getState()==States.CONNECTING)

            {

                _logger.LogInformation("当前状态：CONNECTING。阻塞等待");

                autoResetEvent[].WaitOne();

            }

        }

        public Task<string> CreateZNode(string path,string data,CreateMode createMode,List<ACL> aclList)

        {

            ReConn();

            if(string.IsNullOrEmpty(path)||!path.StartsWith('/'))

            {

                _logger.LogInformation("path路径非法，参数：path：{0}",path);

                return null;

            }

            byte[] dat=new byte[];

            if(string.IsNullOrEmpty(data))

            {

                dat=System.Text.Encoding.Default.GetBytes(data);

            }

            if(createMode==null)

            {

                 _logger.LogInformation("createMode为null,默认使用CreateMode.PERSISTENT");

                createMode=CreateMode.PERSISTENT;

            }

            return _zookeeper.createAsync(path,dat,aclList,createMode);

        }

        public async void Sync(string path)

        {

            try

            {

                _logger.LogInformation("同步成功");

                 await _zookeeper.sync(path);

            }

            catch (Exception ep)

            {

                _logger.LogError("同步失败。", ep);

            }

        }

        public async Task<DataResult> GetDataAsync(string path,Watcher watcher,bool isSync)

        {

            ReConn();

            if(await _zookeeper.existsAsync(path)==null )

            {

                _logger.LogInformation("path不存在");

                return null;

            }

            if (isSync)

            {

                _logger.LogInformation("即将进行同步。");

                try

                {

                   await  _zookeeper.sync(path);

                    _logger.LogInformation("同步成功");

                }

                catch (Exception ep)

                {

                    _logger.LogError("同步失败。", ep);

                }

            }

            return await _zookeeper.getDataAsync(path,watcher);

        }

         public async Task<Stat> SetDataAsync(string path,string data,bool isSync)

        {

            ReConn();

            if(await _zookeeper.existsAsync(path)==null )

            {

                 _logger.LogInformation("path不存在");

                return null;

            }

            byte[] dat=new byte[];

            if(!string.IsNullOrEmpty(data))

            {

                dat=System.Text.Encoding.Default.GetBytes(data);

            }

            return await _zookeeper.setDataAsync(path,dat);

        }

        public async Task<ChildrenResult> GetChildrenAsync(string path, Watcher watcher, bool isSync)

        {

            ReConn();

            if (await _zookeeper.existsAsync(path) == null)

            {

                _logger.LogInformation("path不存在");

                return null;

            }

            if (isSync)

            {

                _logger.LogInformation("即将进行同步。");

                try

                {

                    _logger.LogInformation("开始同步");

                    await _zookeeper.sync(path);

                    _logger.LogInformation("同步成功");

                }

                catch (Exception ep)

                {

                    _logger.LogError("同步失败。", ep);

                }

            }

            return await _zookeeper.getChildrenAsync(path, watcher);

        }

        public async Task DeleteNode(string path)

         {

             ReConn();

              if(await _zookeeper.existsAsync(path)==null )

            {

                 _logger.LogDebug("删除path：path不存在");

                return;

            }

            try

            {

                _logger.LogDebug("删除node：{0}", path);

                await _zookeeper.deleteAsync(path);

            }

            catch (Exception ep)

            {

                _logger.LogError("删除失败", ep);

                return;

            }

        }

        public async Task<bool> SetWatcher(string path,Watcher watcher)

        {

            ReConn();

            var stat = await _zookeeper.existsAsync(path);

            if(stat==null )

            {

                 _logger.LogDebug("判断path是否存在：path不存在");

                return false;

            }

             try

            {

                _logger.LogDebug("设置监控：{0}", path);

                await _zookeeper.getDataAsync(path,watcher);

                return true;

            }

            catch (Exception ep)

            {

                _logger.LogError("设置监控错误", ep);

                return false;

            }

        }

         public  string GetDataByLockNode(string path,string sequenceName,List<ACL> aclList,out string tempNodeOut)

         {

             _logger.LogInformation("获取分布式锁开始。");

             string tempNode=string.Empty;

             tempNodeOut=string.Empty;

            try

            {

                ReConn();

                if (_zookeeper.existsAsync(path).Result == null)

                {

                    _logger.LogDebug("path不存在，创建");

                    CreateZNode(path, "", CreateMode.PERSISTENT, aclList).GetAwaiter().GetResult();

                }

                tempNode = CreateZNode(path + "/" + sequenceName, "", CreateMode.EPHEMERAL_SEQUENTIAL, aclList).Result;

                _logger.LogDebug("创建节点：{0}", tempNode);

                if (tempNode == null)

                {

                    _logger.LogDebug("创建临时序列节点失败。详细参数:path:{0},data:{1},CreateMode:{2}"

                    , path + "/squence", "", CreateMode.EPHEMERAL_SEQUENTIAL);

                    return null;

                }

                _logger.LogInformation("创建成功。");

                // var taskGetData=Task.Run(async () =>{

                //     int circleCount = 0;

                //     while (true)

                //     {

                //         Thread.Sleep(200);

                //         circleCount++;

                //         _logger.LogInformation("循环获取锁。当前循环次数:{0}", circleCount);

                //         try

                //         {

                //             var childList =await GetChildrenAsync(path, null, true);

                //             if (childList == null || childList.Children == null || childList.Children.Count < 1)

                //             {

                //                 _logger.LogWarning("获取子序列失败，计数为零.path:{0}", path);

                //                 return null;

                //             }

                //             _logger.LogInformation("获取path:{0}的子节点：{1}", path, Newtonsoft.Json.JsonConvert.SerializeObject(childList.Children));

                //             var top = childList.Children.OrderBy(or => or).First();

                //             if (path + "/" + top == tempNode)

                //             {

                //                 return tempNode;

                //             }

                //         }

                //         catch (Exception ep)

                //         {

                //             _logger.LogError(ep,"循环获取锁出错。");

                //             return null;

                //         }

                //     }

                // });

                // tempNode = taskGetData.GetAwaiter().GetResult();

                // if (!string.IsNullOrEmpty(tempNode))

                // {

                //     byte[] da = null;

                //     tempNodeOut = tempNode;

                //     da = GetDataAsync(path, null, true).Result.Data;

                //     if (da == null || da.Length < 1)

                //     {

                //         return string.Empty;

                //     }

                //     return System.Text.Encoding.Default.GetString(da);

                // }

                int clycleCount = ;

            GetChild: //这里防止并发出现错误。

                clycleCount++;

                var childList = GetChildrenAsync(path, null, true).GetAwaiter().GetResult();

                if (childList == null || childList.Children == null || childList.Children.Count < )

                {

                    _logger.LogWarning("获取子序列失败，计数为零.path:{0}", path);

                    return null;

                }

                _logger.LogInformation("获取path:{0}的子节点：{1}", path, Newtonsoft.Json.JsonConvert.SerializeObject(childList.Children));

                var top = childList.Children.OrderBy(or => or).First();

                if (path + "/" + top == tempNode)

                {

                    tempNodeOut = tempNode;

                    var da = GetDataAsync(path, null, true).Result.Data;

                    if (da == null || da.Length < )

                    {

                        return string.Empty;

                    }

                    return System.Text.Encoding.Default.GetString(da);

                }

                // bool isSet=

                //     SetWatcher(path + "/" + top,).Result;

                // if(!isSet)

                // {

                //     goto GetChild;

                // }

                bool isSet= SetWatcher(path + "/" + top,new WaitLockWatch(autoResetEvent[], _logger, path,_manualReset)).Result;

                if(!isSet)

                {

                    _logger.LogWarning("没有设置上watcher，需要重新运行一遍。");

                    goto GetChild;

                }

                _manualReset.WaitOne();

                 childList = GetChildrenAsync(path, null, true).GetAwaiter().GetResult();

                if (childList == null || childList.Children == null || childList.Children.Count < )

                {

                    _logger.LogWarning("再次获取子序列失败，计数为零.path:{0}", path);

                    return null;

                }

                _logger.LogInformation("再次获取path:{0}的子节点：{1}", path, Newtonsoft.Json.JsonConvert.SerializeObject(childList.Children));

                top = childList.Children.OrderBy(or => or).First();

                if (path + "/" + top == tempNode)

                {

                    _logger.LogDebug("节点获取到锁权限。");

                    tempNodeOut = tempNode;

                    var da = GetDataAsync(path, null, true).Result.Data;

                    if (da == null || da.Length < )

                    {

                        return string.Empty;

                    }

                    return System.Text.Encoding.Default.GetString(da);

                }

                else

                {

                    _logger.LogDebug("没有获取到锁权限，进行循环。循环第：{0}次",clycleCount);

                    Thread.Sleep();

                    goto GetChild;

                    // Sync(path);

                    //DeleteNode(tempNode).GetAwaiter().GetResult();

                    // DeleteNode(tempNode).GetAwaiter().GetResult();

                    //  _logger.LogError("没有获取到锁，Watcher出现问题，请查看日志。");

                    // if (_zookeeper.existsAsync(tempNode).Result== null)

                    // {

                    //     _logger.LogWarning("tempNode:{0}存在,但是没有获取到锁，在等待的时候，被线程检查程序释放了阻塞，属于误伤"

                    //     ,tempNode);

                    // }

                    // else

                    // {

                    //     _logger.LogError("没有获取到锁，Watcher出现问题，请查看日志。");

                    // }

                }

            }

            catch(ConnectionLossException cle)

            {

                _logger.LogError(cle, "获取同步锁出现错误。连接丢失");

            }

            catch(SessionExpiredException sep)

            {

                _logger.LogError(sep, "获取同步锁出现错误。连接过期");

            }

            catch(KeeperException kep)

            {

                _logger.LogError(kep, "获取同步锁出现错误。操作zookeeper出错");

            }

            catch (Exception ep)

            {

                _logger.LogError(ep, "获取同步锁出现错误。");

                if (!string.IsNullOrEmpty(tempNode))

                {

                    try{

                    DeleteNode(tempNode).GetAwaiter().GetResult();

                    }catch(Exception)

                    {

                    }

                }

            }

            return null;

         }

         private void ReConn()

         {

             _logger.LogInformation("检查连接状态，status:{0}",_zookeeper.getState());

             if(_zookeeper.getState()==States.CLOSED

             ||_zookeeper.getState()== States.NOT_CONNECTED)

             {

                 _logger.LogInformation("连接为关闭，开始重新连接。");

                Conn(_zookeeperOptions);

             }

         }

        public async void Close(string tempNode)

        {

            try

            {

                await _zookeeper.closeAsync();

            }

            catch (Exception ep)

            {

                _logger.LogError("zookeeper关闭失败。", ep);

            }

        }

    }

}

下面看结果

咱们关闭master2，然后看看slave1：

大家看时间上，master2已经2分钟没运行了，如果要有好的监控，还是最好实现remote，也不是什么难事，将通用主机改为webhost，然后写几个api就行了，就会实时的监控quartz主机的状态

，而且分片也会自动去除失败的主机，自动分派任务。咱们看看分片的情况：

master2已经失败，但是这里没有去掉，不过不影响接下来的任务，再者master2的任务如果在失败的时候没执行完成，那么会有一部分数据是没有处理的。

三总结

quartz扩展主要注意两点，一是job listeners和trigger listeners，做好异常处理，尤其是trigger listeners，如果出错，job会失去控制，不激发，而且job状态也会失效，必须重新关闭和重新运行一次。在写业务代码的时候，分片的需要处理的数据源必须是有规律自增的或者是静态的，这样分页才满足业务分片的要求。如果是非自增的或者随机增加，那么quartz就必须把需要处理的主键存进去，但是这样的需求毕竟是少数。

微服务系列的github：https://github.com/ck0074451665/Walt.Framework.git

测试例子：https://github.com/ck0074451665/Walt.MicroServicesTest.git

管理界面：https://pan.baidu.com/s/1gYNDX1j3-XctuPiejV2XPQ

巴特西

asp.net core microservices 架构之分布式自动计算(二)

最新文章

热门文章

巴特西

asp.net core microservices 架构之 分布式自动计算(二)

最新文章

热门文章

asp.net core microservices 架构之分布式自动计算(二)