拆分AntService，为新版本web提供http接口做准备

AntJob.Server/AntService.cs

@@ -2,14 +2,13 @@
 using AntJob.Data;
 using AntJob.Data.Entity;
 using AntJob.Models;
+using AntJob.Server.Services;
 using NewLife;
 using NewLife.Caching;
 using NewLife.Data;
 using NewLife.Log;
 using NewLife.Net;
 using NewLife.Remoting;
-using NewLife.Security;
-using NewLife.Serialization;
 using NewLife.Threading;
 using XCode;
 
@@ -31,14 +30,18 @@ class AntService : IApi, IActionFilter
 
     private App _App;
     private INetSession _Net;
+    private readonly AppService _appService;
+    private readonly JobService _jobService;
     private readonly ICacheProvider _cacheProvider;
     private readonly ITracer _tracer;
     private readonly ILog _log;
     #endregion
 
     #region 构造
-    public AntService(ICacheProvider cacheProvider, ITracer tracer, ILog log)
+    public AntService(AppService appService, JobService jobService, ICacheProvider cacheProvider, ITracer tracer, ILog log)
     {
+        _appService = appService;
+        _jobService = jobService;
         _cacheProvider = cacheProvider;
         _tracer = tracer;
         _log = log;
@@ -93,104 +96,25 @@ class AntService : IApi, IActionFilter
     {
         if (model.User.IsNullOrEmpty()) throw new ArgumentNullException(nameof(model.User));
 
-        _log.Info("[{0}]从[{1}]登录[{2}@{3}]", model.User, _Net.Remote, model.Machine, model.ProcessId);
-
-        // 找应用
-        var autoReg = false;
-        var app = App.FindByName(model.User);
-        if (app == null || app.Secret.MD5() != model.Pass)
-        {
-            app = CheckApp(app, model.User, model.Pass, _Net.Remote.Host);
-            if (app == null) throw new ArgumentOutOfRangeException(nameof(model.User));
-
-            autoReg = true;
-        }
-
-        if (app == null) throw new Exception($"应用[{model.User}]不存在！");
-        if (!app.Enable) throw new Exception("已禁用！");
-
-        // 核对密码
-        if (!autoReg && !app.Secret.IsNullOrEmpty())
-        {
-            var pass2 = app.Secret.MD5();
-            if (model.Pass != pass2) throw new Exception("密码错误！");
-        }
-
-        // 版本和编译时间
-        if (app.Version.IsNullOrEmpty() || app.Version.CompareTo(model.Version) < 0) app.Version = model.Version;
-        if (app.CompileTime < model.Compile) app.CompileTime = model.Compile;
-        if (app.DisplayName.IsNullOrEmpty()) app.DisplayName = model.DisplayName;
-
-        app.Save();
-
-        // 应用上线
-        var online = CreateOnline(app, _Net, model.Machine, model.ProcessId);
-        online.Version = model.Version;
-        online.CompileTime = model.Compile;
-        online.Save();
+        var (app, rs) = _appService.Login(model, _Net.Remote.Host);
 
         // 记录当前用户
         Session["App"] = app;
 
-        WriteHistory(app, autoReg ? "注册" : "登录", true, $"[{model.User}/{model.Pass}]在[{model.Machine}@{model.ProcessId}]登录[{app}]成功");
-
-        var rs = new LoginResponse { Name = app.Name, DisplayName = app.DisplayName };
-        if (autoReg) rs.Secret = app.Secret;
-
         return rs;
     }
 
-    protected virtual App CheckApp(App app, String user, String pass, String ip)
-    {
-        //  本地账号不存在时
-        var name = user;
-        if (app == null)
-        {
-            // 是否支持自动注册
-            var set = Setting.Current;
-            if (!set.AutoRegistry) throw new Exception($"找不到应用[{name}]");
-
-            app = new App();
-            app.Secret = Rand.NextString(16);
-        }
-        else if (app.Secret.MD5() != pass)
-        {
-            // 是否支持自动注册
-            var set = Setting.Current;
-            if (!set.AutoRegistry) throw new Exception($"应用[{name}]申请重新激活，但服务器设置禁止自动注册");
-
-            if (app.Secret.IsNullOrEmpty()) app.Secret = Rand.NextString(16);
-        }
-
-        if (app.ID == 0)
-        {
-            app.Name = name;
-            app.CreateIP = ip;
-            app.CreateTime = DateTime.Now;
-
-            // 首次打开
-            app.Enable = true;
-        }
-
-        app.UpdateIP = ip;
-        app.UpdateTime = DateTime.Now;
-
-        //app.Save();
-
-        return app;
-    }
+    /// <summary>获取当前应用的所有在线实例</summary>
+    /// <returns></returns>
+    [Api(nameof(GetPeers))]
+    public PeerModel[] GetPeers() => _appService.GetPeers(_App);
     #endregion
 
     #region 业务
     /// <summary>获取指定名称的作业</summary>
     /// <returns></returns>
     [Api(nameof(GetJobs))]
-    public IJob[] GetJobs()
-    {
-        var jobs = Job.FindAllByAppID(_App.ID);
-
-        return jobs.Select(e => e.ToModel()).ToArray();
-    }
+    public IJob[] GetJobs() => _jobService.GetJobs(_App);
 
     /// <summary>批量添加作业</summary>
     /// <param name="jobs"></param>
@@ -200,44 +124,7 @@ class AntService : IApi, IActionFilter
     {
         if (jobs == null || jobs.Length == 0) return new String[0];
 
-        var myJobs = Job.FindAllByAppID(_App.ID);
-        var list = new List<String>();
-        foreach (var item in jobs)
-        {
-            var jb = myJobs.FirstOrDefault(e => e.Name.EqualIgnoreCase(item.Name));
-            jb ??= new Job
-            {
-                AppID = _App.ID,
-                Name = item.Name,
-                Enable = item.Enable,
-                Start = item.Start,
-                End = item.End,
-                Offset = item.Offset,
-                Step = item.Step,
-                BatchSize = item.BatchSize,
-                MaxTask = item.MaxTask,
-                Mode = item.Mode,
-                MaxError = 100,
-            };
-
-            if (item.Mode > 0) jb.Mode = item.Mode;
-            if (!item.DisplayName.IsNullOrEmpty()) jb.DisplayName = item.DisplayName;
-            if (!item.Description.IsNullOrEmpty()) jb.Remark = item.Description;
-            if (!item.ClassName.IsNullOrEmpty()) jb.ClassName = item.ClassName;
-            if (jb.Topic.IsNullOrEmpty()) jb.Topic = item.Topic;
-
-            if (jb.Save() != 0)
-            {
-                _log.Info("[{0}]更新作业[{1}] @[{2}]", _App, item.Name, _Net.Remote);
-
-                // 更新作业数
-                jb.SaveAsync();
-
-                list.Add(jb.Name);
-            }
-        }
-
-        return list.ToArray();
+        return _jobService.AddJobs(_App, jobs);
     }
 
     /// <summary>申请作业任务</summary>
@@ -249,70 +136,7 @@ class AntService : IApi, IActionFilter
         var job = model.Job?.Trim();
         if (job.IsNullOrEmpty()) return new TaskModel[0];
 
-        var app = _App;
-        if (app == null) return new TaskModel[0];
-
-        // 应用停止发放作业
-        app = App.FindByID(app.ID) ?? app;
-        if (!app.Enable) return new TaskModel[0];
-        var jb = app.Jobs.FirstOrDefault(e => e.Name == job);
-
-        // 全局锁，确保单个作业只有一个线程在分配作业
-        using var ck = _cacheProvider.AcquireLock($"antjob:lock:{jb.ID}", 15_000);
-
-        // 找到作业。为了确保能够快速拿到新的作业参数，这里做二次查询
-        if (jb != null)
-            jb = Job.Find(Job._.ID == jb.ID);
-        else
-            jb = Job.FindByAppIDAndName(app.ID, job);
-
-        if (jb == null) throw new XException($"应用[{app.ID}/{app.Name}]下未找到作业[{job}]");
-        if (jb.Step == 0 || jb.Start.Year <= 2000) throw new XException("作业[{0}/{1}]未设置开始时间或步进", jb.ID, jb.Name);
-
-        var online = GetOnline(app, _Net);
-
-        var list = new List<JobTask>();
-
-        // 每分钟检查一下错误任务和中断任务
-        CheckErrorTask(app, jb, model.Count, list);
-
-        // 错误项不够时，增加切片
-        if (list.Count < model.Count)
-        {
-            //var ps = ControllerContext.Current.Parameters;
-            var server = online.Name;
-            var pid = online.ProcessId;
-            //var topic = ps["topic"] + "";
-            var ip = _Net.Remote.Host;
-
-            switch (jb.Mode)
-            {
-                case JobModes.Message:
-                    list.AddRange(jb.AcquireMessage(model.Topic, server, ip, pid, model.Count - list.Count, _cacheProvider.Cache));
-                    break;
-                case JobModes.Data:
-                case JobModes.Alarm:
-                //case JobModes.CSharp:
-                //case JobModes.Sql:
-                default:
-                    {
-                        // 如果能够切片，则查询数据库后进入，避免缓存导致重复
-                        if (jb.TrySplit(jb.Start, jb.Step, out var end))
-                        {
-                            // 申请任务前，不能再查数据库，那样子会导致多线程脏读，从而出现多客户端分到相同任务的情况
-                            //jb = Job.FindByKey(jb.ID);
-                            list.AddRange(jb.Acquire(server, ip, pid, model.Count - list.Count, _cacheProvider.Cache));
-                        }
-                    }
-                    break;
-            }
-        }
-
-        // 记录状态
-        online.Tasks += list.Count;
-        online.SaveAsync();
-
-        return list.Select(e => e.ToModel()).ToArray();
+        return _jobService.Acquire(_App, model);
     }
 
     private void CheckErrorTask(App app, Job jb, Int32 count, List<JobTask> list)
@@ -356,84 +180,9 @@ class AntService : IApi, IActionFilter
         var messages = model?.Messages?.Where(e => !e.IsNullOrEmpty()).Distinct().ToArray();
         if (messages == null || messages.Length == 0) return 0;
 
-        var app = _App;
-
-        // 去重过滤
-        if (model.Unique)
-        {
-            // 如果消息较短，采用缓存做去重过滤
-            if (messages.All(e => e.Length < 64))
-            {
-                var msgs = new List<String>();
-                foreach (var item in messages)
-                {
-                    var key = $"antjob:{app.ID}:{model.Topic}:{item}";
-                    if (_cacheProvider.Cache.Add(key, item, 2 * 3600)) msgs.Add(item);
-                }
-                messages = msgs.ToArray();
-            }
-            else
-            {
-                messages = AppMessage.Filter(app.ID, model.Topic, messages);
-                if (messages.Length == 0) return 0;
-            }
-        }
-
-        var ms = new List<AppMessage>();
-
-        var total = 0;
-        var now = DateTime.Now;
-        // 延迟需要基于任务开始时间，而不能用使用当前时间，防止回头跑数据时无法快速执行
-        var dTime = now.AddSeconds(model.DelayTime);
-
-        var jb = Job.FindByAppIDAndName(app.ID, model.Job);
-        var snow = AppMessage.Meta.Factory.Snow;
-        foreach (var item in messages)
-        {
-            var jm = new AppMessage
-            {
-                Id = snow.NewId(),
-                AppID = app.ID,
-                JobID = jb == null ? 0 : jb.ID,
-                Topic = model.Topic,
-                Data = item,
-            };
-
-            jm.CreateTime = jm.UpdateTime = now;
-
-            // 雪花Id直接指定消息在未来的消费时间
-            if (model.DelayTime > 0)
-            {
-                jm.Id = snow.NewId(dTime);
-                jm.UpdateTime = dTime;
-            }
-
-            ms.Add(jm);
-        }
-
-        // 记录消息积压数
-        total = ms.BatchInsert();
-
-        // 增加消息数
-        if (total < 0) total = messages.Length;
-        if (total > 0)
-        {
-            var job2 = app.Jobs?.FirstOrDefault(e => e.Topic == model.Topic);
-            if (job2 != null)
-            {
-                job2.MessageCount += total;
-                job2.SaveAsync();
-            }
-
-            app.MessageCount += total;
-            app.SaveAsync();
-        }
-
-        return total;
+        return _jobService.Produce(_App, model);
     }
-    #endregion
 
-    #region 状态报告
     /// <summary>报告状态（进度、成功、错误）</summary>
     /// <param name="task"></param>
     /// <returns></returns>
@@ -442,190 +191,7 @@ class AntService : IApi, IActionFilter
     {
         if (task == null || task.ID == 0) throw new InvalidOperationException("无效操作 TaskID=" + task?.ID);
 
-        // 判断是否有权
-        var app = _App;
-
-        var jt = JobTask.FindByID(task.ID) ?? throw new InvalidOperationException($"找不到任务[{task.ID}]");
-        var job = Job.FindByID(jt.JobID);
-        if (job == null || job.AppID != app.ID)
-        {
-            _log.Info(task.ToJson());
-            throw new InvalidOperationException($"应用[{app}]无权操作作业[{job}#{jt}]");
-        }
-
-        // 只有部分字段允许客户端修改
-        if (task.Status > 0) jt.Status = task.Status;
-
-        jt.Speed = task.Speed;
-        jt.Total = task.Total;
-        jt.Success = task.Success;
-        jt.Cost = task.Cost;
-        jt.Key = task.Key;
-        jt.Message = task.Message;
-
-        // 已终结的作业，汇总统计
-        if (task.Status == JobStatus.完成 || task.Status == JobStatus.错误)
-        {
-            jt.Times++;
-
-            SetJobFinish(job, jt);
-
-            // 记录状态
-            UpdateOnline(app, jt, _Net);
-        }
-        if (task.Status == JobStatus.错误)
-        {
-            SetJobError(job, jt);
-
-            jt.Error++;
-            //ji.Message = err.Message;
-
-            // 出错时判断如果超过最大错误数，则停止作业
-            CheckMaxError(app, job);
-        }
-
-        // 从创建到完成的全部耗时
-        var ts = DateTime.Now - jt.CreateTime;
-        jt.FullCost = (Int32)ts.TotalSeconds;
-
-        jt.SaveAsync();
-        //ji.Save();
-
-        return true;
+        return _jobService.Report(_App, task);
     }
-
-    private void SetJobFinish(Job job, JobTask task)
-    {
-        job.Total += task.Total;
-        job.Success += task.Success;
-        job.Error += task.Error;
-        job.Times++;
-
-        var ths = job.MaxTask;
-
-        var p1 = task.Speed * ths;
-
-        if (p1 > 0)
-        {
-            // 平均速度
-            if (job.Speed > 0)
-                job.Speed = (Int32)((job.Speed * 3L + p1) / 4);
-            else
-                job.Speed = p1;
-        }
-
-        job.SaveAsync();
-        //job.Save();
-    }
-
-    private JobError SetJobError(Job job, JobTask task)
-    {
-        var err = new JobError
-        {
-            AppID = job.AppID,
-            JobID = job.ID,
-            TaskID = task.ID,
-            Start = task.Start,
-            End = task.End,
-            Data = task.Data,
-
-            Server = task.Server,
-            ProcessID = task.ProcessID,
-            Client = task.Client,
-
-            CreateTime = DateTime.Now,
-            UpdateTime = DateTime.Now,
-        };
-
-        var msg = task.Message;
-        if (!msg.IsNullOrEmpty() && msg.Contains("Exception:")) msg = msg.Substring("Exception:").Trim();
-        err.Message = msg;
-
-        err.Insert();
-
-        return err;
-    }
-
-    private void CheckMaxError(App app, Job job)
-    {
-        // 出错时判断如果超过最大错误数，则停止作业
-        var maxError = job.MaxError < 1 ? 100 : job.MaxError;
-        if (job.Enable && job.Error > maxError)
-        {
-            job.MaxError = maxError;
-            job.Enable = false;
-
-            //job.SaveAsync();
-            (job as IEntity).Update();
-        }
-    }
-    #endregion
-
-    #region 在线状态
-    /// <summary>获取当前应用的所有在线实例</summary>
-    /// <returns></returns>
-    [Api(nameof(GetPeers))]
-    public PeerModel[] GetPeers()
-    {
-        var olts = AppOnline.FindAllByAppID(_App.ID);
-
-        return olts.Select(e => e.ToModel()).ToArray();
-    }
-
-    AppOnline CreateOnline(App app, INetSession ns, String machine, Int32 pid)
-    {
-        var ip = ns.Remote.Host;
-
-        var online = GetOnline(app, ns);
-        online.Client = $"{(ip.IsNullOrEmpty() ? machine : ip)}@{pid}";
-        online.Name = machine;
-        online.ProcessId = pid;
-        online.UpdateIP = ip;
-        //online.Version = version;
-
-        online.Server = Local + "";
-        //online.Save();
-
-        // 真正的用户
-        Session["AppOnline"] = online;
-
-        // 下线
-        ns.OnDisposed += (s, e) =>
-        {
-            online.Delete();
-            WriteHistory(online.App, "下线", true, $"[{online.Name}]登录于{online.CreateTime}，最后活跃于{online.UpdateTime}");
-        };
-
-        return online;
-    }
-
-    AppOnline GetOnline(App app, INetSession ns)
-    {
-        if (Session["AppOnline"] is AppOnline online) return online;
-
-        var ip = ns.Remote.Host;
-        var ins = ns.Remote.EndPoint + "";
-        online = AppOnline.FindByInstance(ins) ?? new AppOnline { CreateIP = ip };
-        online.AppID = app.ID;
-        online.Instance = ins;
-
-        return online;
-    }
-
-    void UpdateOnline(App app, JobTask ji, INetSession ns)
-    {
-        var online = GetOnline(app, ns);
-        online.Total += ji.Total;
-        online.Success += ji.Success;
-        online.Error += ji.Error;
-        online.Cost += ji.Cost;
-        online.Speed = ji.Speed;
-        online.LastKey = ji.Key;
-        online.SaveAsync();
-    }
-    #endregion
-
-    #region 写历史
-    void WriteHistory(App app, String action, Boolean success, String remark) => AppHistory.Create(app ?? _App, action, success, remark, Local + "", _Net.Remote?.Host);
     #endregion
 }

AntJob.Server/Program.cs

@@ -1,4 +1,5 @@
 using AntJob.Server;
+using AntJob.Server.Services;
 using NewLife;
 using NewLife.Caching;
 using NewLife.Caching.Services;
@@ -34,6 +35,8 @@ if (set2.IsNew)
 // 分布式缓存，锚定配置中心RedisCache，若无配置则使用本地MemoryCache
 // 集群部署时，务必使用RedisCache，内部将使用Redis实现分布式锁
 services.AddSingleton<ICacheProvider, RedisCacheProvider>();
+services.AddSingleton<AppService>();
+services.AddSingleton<JobService>();
 
 // 预热数据层，执行反向工程建表等操作
 EntityFactory.InitConnection("Ant");

AntJob.Server/Services/AppService.cs

@@ -0,0 +1,192 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using AntJob.Data;
+using AntJob.Data.Entity;
+using NewLife;
+using NewLife.Caching;
+using NewLife.Log;
+using NewLife.Net;
+using NewLife.Security;
+using AntJob.Models;
+
+namespace AntJob.Server.Services;
+
+public class AppService
+{
+    private readonly ICacheProvider _cacheProvider;
+    private readonly ITracer _tracer;
+    private readonly ILog _log;
+
+    public AppService(ICacheProvider cacheProvider, ITracer tracer, ILog log)
+    {
+        _cacheProvider = cacheProvider;
+        _tracer = tracer;
+        _log = log;
+    }
+
+    #region 登录
+    /// <summary>应用登录</summary>
+    /// <param name="model">模型</param>
+    /// <returns></returns>
+    public (App, LoginResponse) Login(LoginModel model, String ip)
+    {
+        if (model.User.IsNullOrEmpty()) throw new ArgumentNullException(nameof(model.User));
+
+        _log.Info("[{0}]从[{1}]登录[{2}@{3}]", model.User, ip, model.Machine, model.ProcessId);
+
+        // 找应用
+        var autoReg = false;
+        var app = App.FindByName(model.User);
+        if (app == null || app.Secret.MD5() != model.Pass)
+        {
+            app = CheckApp(app, model.User, model.Pass, ip);
+            if (app == null) throw new ArgumentOutOfRangeException(nameof(model.User));
+
+            autoReg = true;
+        }
+
+        if (app == null) throw new Exception($"应用[{model.User}]不存在！");
+        if (!app.Enable) throw new Exception("已禁用！");
+
+        // 核对密码
+        if (!autoReg && !app.Secret.IsNullOrEmpty())
+        {
+            var pass2 = app.Secret.MD5();
+            if (model.Pass != pass2) throw new Exception("密码错误！");
+        }
+
+        // 版本和编译时间
+        if (app.Version.IsNullOrEmpty() || app.Version.CompareTo(model.Version) < 0) app.Version = model.Version;
+        if (app.CompileTime < model.Compile) app.CompileTime = model.Compile;
+        if (app.DisplayName.IsNullOrEmpty()) app.DisplayName = model.DisplayName;
+
+        app.Save();
+
+        // 应用上线
+        var online = CreateOnline(app, _Net, model.Machine, model.ProcessId);
+        online.Version = model.Version;
+        online.CompileTime = model.Compile;
+        online.Save();
+
+        //// 记录当前用户
+        //Session["App"] = app;
+
+        WriteHistory(app, autoReg ? "注册" : "登录", true, $"[{model.User}/{model.Pass}]在[{model.Machine}@{model.ProcessId}]登录[{app}]成功");
+
+        var rs = new LoginResponse { Name = app.Name, DisplayName = app.DisplayName };
+        if (autoReg) rs.Secret = app.Secret;
+
+        return (app, rs);
+    }
+
+    protected virtual App CheckApp(App app, String user, String pass, String ip)
+    {
+        //  本地账号不存在时
+        var name = user;
+        if (app == null)
+        {
+            // 是否支持自动注册
+            var set = AntJobSetting.Current;
+            if (!set.AutoRegistry) throw new Exception($"找不到应用[{name}]");
+
+            app = new App
+            {
+                Secret = Rand.NextString(16)
+            };
+        }
+        else if (app.Secret.MD5() != pass)
+        {
+            // 是否支持自动注册
+            var set = AntJobSetting.Current;
+            if (!set.AutoRegistry) throw new Exception($"应用[{name}]申请重新激活，但服务器设置禁止自动注册");
+
+            if (app.Secret.IsNullOrEmpty()) app.Secret = Rand.NextString(16);
+        }
+
+        if (app.ID == 0)
+        {
+            app.Name = name;
+            app.CreateIP = ip;
+            app.CreateTime = DateTime.Now;
+
+            // 首次打开
+            app.Enable = true;
+        }
+
+        app.UpdateIP = ip;
+        app.UpdateTime = DateTime.Now;
+
+        //app.Save();
+
+        return app;
+    }
+    #endregion
+
+    #region 在线状态
+    /// <summary>获取当前应用的所有在线实例</summary>
+    /// <returns></returns>
+    public PeerModel[] GetPeers(App app)
+    {
+        var olts = AppOnline.FindAllByAppID(app.ID);
+
+        return olts.Select(e => e.ToModel()).ToArray();
+    }
+
+    AppOnline CreateOnline(App app, INetSession ns, String machine, Int32 pid)
+    {
+        var ip = ns.Remote.Host;
+
+        var online = GetOnline(app, ns);
+        online.Client = $"{(ip.IsNullOrEmpty() ? machine : ip)}@{pid}";
+        online.Name = machine;
+        online.ProcessId = pid;
+        online.UpdateIP = ip;
+        //online.Version = version;
+
+        online.Server = Local + "";
+        //online.Save();
+
+        // 真正的用户
+        Session["AppOnline"] = online;
+
+        // 下线
+        ns.OnDisposed += (s, e) =>
+        {
+            online.Delete();
+            WriteHistory(online.App, "下线", true, $"[{online.Name}]登录于{online.CreateTime}，最后活跃于{online.UpdateTime}");
+        };
+
+        return online;
+    }
+
+    public AppOnline GetOnline(App app, INetSession ns)
+    {
+        if (Session["AppOnline"] is AppOnline online) return online;
+
+        var ip = ns.Remote.Host;
+        var ins = ns.Remote.EndPoint + "";
+        online = AppOnline.FindByInstance(ins) ?? new AppOnline { CreateIP = ip };
+        online.AppID = app.ID;
+        online.Instance = ins;
+
+        return online;
+    }
+
+    public void UpdateOnline(App app, JobTask ji, INetSession ns)
+    {
+        var online = GetOnline(app, ns);
+        online.Total += ji.Total;
+        online.Success += ji.Success;
+        online.Error += ji.Error;
+        online.Cost += ji.Cost;
+        online.Speed = ji.Speed;
+        online.LastKey = ji.Key;
+        online.SaveAsync();
+    }
+    #endregion
+
+    #region 写历史
+    public void WriteHistory(App app, String action, Boolean success, String remark) => AppHistory.Create(app, action, success, remark, Local + "", _Net.Remote?.Host);
+    #endregion
+}

AntJob.Server/Services/JobService.cs

@@ -0,0 +1,400 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using AntJob.Data;
+using AntJob.Data.Entity;
+using AntJob.Models;
+using NewLife;
+using NewLife.Caching;
+using NewLife.Data;
+using NewLife.Log;
+using NewLife.Serialization;
+using NewLife.Threading;
+using XCode;
+
+namespace AntJob.Server.Services;
+
+public class JobService
+{
+    private readonly AppService _appService;
+    private readonly ICacheProvider _cacheProvider;
+    private readonly ITracer _tracer;
+    private readonly ILog _log;
+
+    public JobService(AppService appService, ICacheProvider cacheProvider, ITracer tracer, ILog log)
+    {
+        _appService = appService;
+        _cacheProvider = cacheProvider;
+        _tracer = tracer;
+        _log = log;
+    }
+
+    #region 业务
+    /// <summary>获取指定名称的作业</summary>
+    /// <returns></returns>
+    public IJob[] GetJobs(App app)
+    {
+        var jobs = Job.FindAllByAppID(app.ID);
+
+        return jobs.Select(e => e.ToModel()).ToArray();
+    }
+
+    /// <summary>批量添加作业</summary>
+    /// <param name="jobs"></param>
+    /// <returns></returns>
+    public String[] AddJobs(App app, JobModel[] jobs)
+    {
+        if (jobs == null || jobs.Length == 0) return new String[0];
+
+        var myJobs = Job.FindAllByAppID(app.ID);
+        var list = new List<String>();
+        foreach (var item in jobs)
+        {
+            var jb = myJobs.FirstOrDefault(e => e.Name.EqualIgnoreCase(item.Name));
+            jb ??= new Job
+            {
+                AppID = app.ID,
+                Name = item.Name,
+                Enable = item.Enable,
+                Start = item.Start,
+                End = item.End,
+                Offset = item.Offset,
+                Step = item.Step,
+                BatchSize = item.BatchSize,
+                MaxTask = item.MaxTask,
+                Mode = item.Mode,
+                MaxError = 100,
+            };
+
+            if (item.Mode > 0) jb.Mode = item.Mode;
+            if (!item.DisplayName.IsNullOrEmpty()) jb.DisplayName = item.DisplayName;
+            if (!item.Description.IsNullOrEmpty()) jb.Remark = item.Description;
+            if (!item.ClassName.IsNullOrEmpty()) jb.ClassName = item.ClassName;
+            if (jb.Topic.IsNullOrEmpty()) jb.Topic = item.Topic;
+
+            if (jb.Save() != 0)
+            {
+                // 更新作业数
+                jb.SaveAsync();
+
+                list.Add(jb.Name);
+            }
+        }
+
+        return list.ToArray();
+    }
+
+    /// <summary>申请作业任务</summary>
+    /// <param name="model">模型</param>
+    /// <returns></returns>
+    public ITask[] Acquire(App app, AcquireModel model)
+    {
+        var job = model.Job?.Trim();
+        if (job.IsNullOrEmpty()) return new TaskModel[0];
+
+        if (app == null) return new TaskModel[0];
+
+        // 应用停止发放作业
+        app = App.FindByID(app.ID) ?? app;
+        if (!app.Enable) return new TaskModel[0];
+        var jb = app.Jobs.FirstOrDefault(e => e.Name == job);
+
+        // 全局锁，确保单个作业只有一个线程在分配作业
+        using var ck = _cacheProvider.AcquireLock($"antjob:lock:{jb.ID}", 15_000);
+
+        // 找到作业。为了确保能够快速拿到新的作业参数，这里做二次查询
+        if (jb != null)
+            jb = Job.Find(Job._.ID == jb.ID);
+        else
+            jb = Job.FindByAppIDAndName(app.ID, job);
+
+        if (jb == null) throw new XException($"应用[{app.ID}/{app.Name}]下未找到作业[{job}]");
+        if (jb.Step == 0 || jb.Start.Year <= 2000) throw new XException("作业[{0}/{1}]未设置开始时间或步进", jb.ID, jb.Name);
+
+        var online = _appService.GetOnline(app, _Net);
+
+        var list = new List<JobTask>();
+
+        // 每分钟检查一下错误任务和中断任务
+        CheckErrorTask(app, jb, model.Count, list);
+
+        // 错误项不够时，增加切片
+        if (list.Count < model.Count)
+        {
+            //var ps = ControllerContext.Current.Parameters;
+            var server = online.Name;
+            var pid = online.ProcessId;
+            //var topic = ps["topic"] + "";
+            var ip = _Net.Remote.Host;
+
+            switch (jb.Mode)
+            {
+                case JobModes.Message:
+                    list.AddRange(jb.AcquireMessage(model.Topic, server, ip, pid, model.Count - list.Count, _cacheProvider.Cache));
+                    break;
+                case JobModes.Data:
+                case JobModes.Alarm:
+                //case JobModes.CSharp:
+                //case JobModes.Sql:
+                default:
+                    {
+                        // 如果能够切片，则查询数据库后进入，避免缓存导致重复
+                        if (jb.TrySplit(jb.Start, jb.Step, out var end))
+                        {
+                            // 申请任务前，不能再查数据库，那样子会导致多线程脏读，从而出现多客户端分到相同任务的情况
+                            //jb = Job.FindByKey(jb.ID);
+                            list.AddRange(jb.Acquire(server, ip, pid, model.Count - list.Count, _cacheProvider.Cache));
+                        }
+                    }
+                    break;
+            }
+        }
+
+        // 记录状态
+        online.Tasks += list.Count;
+        online.SaveAsync();
+
+        return list.Select(e => e.ToModel()).ToArray();
+    }
+
+    private void CheckErrorTask(App app, Job jb, Int32 count, List<JobTask> list)
+    {
+        // 每分钟检查一下错误任务和中断任务
+        var nextKey = $"_NextAcquireOld_{jb.ID}";
+        var now = TimerX.Now;
+        var ext = Session as IExtend;
+        var next = (DateTime)(ext[nextKey] ?? DateTime.MinValue);
+        if (next < now)
+        {
+            //var ps = ControllerContext.Current.Parameters;
+            //var server = ps["server"] + "";
+            //var pid = ps["pid"].ToInt();
+            var online = _appService.GetOnline(app, _Net);
+            var ip = _Net.Remote.Host;
+
+            next = now.AddSeconds(60);
+            list.AddRange(jb.AcquireOld(online.Server, ip, online.ProcessId, count, _cacheProvider.Cache));
+
+            if (list.Count > 0)
+            {
+                // 既然有数据，待会还来
+                next = now;
+
+                var n1 = list.Count(e => e.Status == JobStatus.错误 || e.Status == JobStatus.取消);
+                var n2 = list.Count(e => e.Status == JobStatus.就绪 || e.Status == JobStatus.抽取中 || e.Status == JobStatus.处理中);
+                _log.Info("作业[{0}/{1}]准备处理[{2}]个错误和[{3}]超时任务 [{4}]", app, jb.Name, n1, n2, list.Join(",", e => e.ID + ""));
+            }
+            else
+                ext[nextKey] = next;
+        }
+    }
+
+    /// <summary>生产消息</summary>
+    /// <param name="model">模型</param>
+    /// <returns></returns>
+    public Int32 Produce(App app, ProduceModel model)
+    {
+        var messages = model?.Messages?.Where(e => !e.IsNullOrEmpty()).Distinct().ToArray();
+        if (messages == null || messages.Length == 0) return 0;
+
+        // 去重过滤
+        if (model.Unique)
+        {
+            // 如果消息较短，采用缓存做去重过滤
+            if (messages.All(e => e.Length < 64))
+            {
+                var msgs = new List<String>();
+                foreach (var item in messages)
+                {
+                    var key = $"antjob:{app.ID}:{model.Topic}:{item}";
+                    if (_cacheProvider.Cache.Add(key, item, 2 * 3600)) msgs.Add(item);
+                }
+                messages = msgs.ToArray();
+            }
+            else
+            {
+                messages = AppMessage.Filter(app.ID, model.Topic, messages);
+                if (messages.Length == 0) return 0;
+            }
+        }
+
+        var ms = new List<AppMessage>();
+
+        var total = 0;
+        var now = DateTime.Now;
+        // 延迟需要基于任务开始时间，而不能用使用当前时间，防止回头跑数据时无法快速执行
+        var dTime = now.AddSeconds(model.DelayTime);
+
+        var jb = Job.FindByAppIDAndName(app.ID, model.Job);
+        var snow = AppMessage.Meta.Factory.Snow;
+        foreach (var item in messages)
+        {
+            var jm = new AppMessage
+            {
+                Id = snow.NewId(),
+                AppID = app.ID,
+                JobID = jb == null ? 0 : jb.ID,
+                Topic = model.Topic,
+                Data = item,
+            };
+
+            jm.CreateTime = jm.UpdateTime = now;
+
+            // 雪花Id直接指定消息在未来的消费时间
+            if (model.DelayTime > 0)
+            {
+                jm.Id = snow.NewId(dTime);
+                jm.UpdateTime = dTime;
+            }
+
+            ms.Add(jm);
+        }
+
+        // 记录消息积压数
+        total = ms.Insert();
+
+        // 增加消息数
+        if (total < 0) total = messages.Length;
+        if (total > 0)
+        {
+            var job2 = app.Jobs?.FirstOrDefault(e => e.Topic == model.Topic);
+            if (job2 != null)
+            {
+                job2.MessageCount += total;
+                job2.SaveAsync();
+            }
+
+            app.MessageCount += total;
+            app.SaveAsync();
+        }
+
+        return total;
+    }
+    #endregion
+
+    #region 状态报告
+    /// <summary>报告状态（进度、成功、错误）</summary>
+    /// <param name="task"></param>
+    /// <returns></returns>
+    public Boolean Report(App app, TaskResult task)
+    {
+        if (task == null || task.ID == 0) throw new InvalidOperationException("无效操作 TaskID=" + task?.ID);
+
+        // 判断是否有权
+
+        var jt = JobTask.FindByID(task.ID) ?? throw new InvalidOperationException($"找不到任务[{task.ID}]");
+        var job = Job.FindByID(jt.JobID);
+        if (job == null || job.AppID != app.ID)
+        {
+            _log.Info(task.ToJson());
+            throw new InvalidOperationException($"应用[{app}]无权操作作业[{job}#{jt}]");
+        }
+
+        // 只有部分字段允许客户端修改
+        if (task.Status > 0) jt.Status = task.Status;
+
+        jt.Speed = task.Speed;
+        jt.Total = task.Total;
+        jt.Success = task.Success;
+        jt.Cost = task.Cost;
+        jt.Key = task.Key;
+        jt.Message = task.Message;
+
+        // 已终结的作业，汇总统计
+        if (task.Status == JobStatus.完成 || task.Status == JobStatus.错误)
+        {
+            jt.Times++;
+
+            SetJobFinish(job, jt);
+
+            // 记录状态
+            _appService.UpdateOnline(app, jt, _Net);
+        }
+        if (task.Status == JobStatus.错误)
+        {
+            SetJobError(job, jt);
+
+            jt.Error++;
+            //ji.Message = err.Message;
+
+            // 出错时判断如果超过最大错误数，则停止作业
+            CheckMaxError(app, job);
+        }
+
+        // 从创建到完成的全部耗时
+        var ts = DateTime.Now - jt.CreateTime;
+        jt.FullCost = (Int32)ts.TotalSeconds;
+
+        jt.SaveAsync();
+        //ji.Save();
+
+        return true;
+    }
+
+    private void SetJobFinish(Job job, JobTask task)
+    {
+        job.Total += task.Total;
+        job.Success += task.Success;
+        job.Error += task.Error;
+        job.Times++;
+
+        var ths = job.MaxTask;
+
+        var p1 = task.Speed * ths;
+
+        if (p1 > 0)
+        {
+            // 平均速度
+            if (job.Speed > 0)
+                job.Speed = (Int32)((job.Speed * 3L + p1) / 4);
+            else
+                job.Speed = p1;
+        }
+
+        job.SaveAsync();
+        //job.Save();
+    }
+
+    private JobError SetJobError(Job job, JobTask task)
+    {
+        var err = new JobError
+        {
+            AppID = job.AppID,
+            JobID = job.ID,
+            TaskID = task.ID,
+            Start = task.Start,
+            End = task.End,
+            Data = task.Data,
+
+            Server = task.Server,
+            ProcessID = task.ProcessID,
+            Client = task.Client,
+
+            CreateTime = DateTime.Now,
+            UpdateTime = DateTime.Now,
+        };
+
+        var msg = task.Message;
+        if (!msg.IsNullOrEmpty() && msg.Contains("Exception:")) msg = msg.Substring("Exception:").Trim();
+        err.Message = msg;
+
+        err.Insert();
+
+        return err;
+    }
+
+    private void CheckMaxError(App app, Job job)
+    {
+        // 出错时判断如果超过最大错误数，则停止作业
+        var maxError = job.MaxError < 1 ? 100 : job.MaxError;
+        if (job.Enable && job.Error > maxError)
+        {
+            job.MaxError = maxError;
+            job.Enable = false;
+
+            //job.SaveAsync();
+            (job as IEntity).Update();
+        }
+    }
+    #endregion
+}

AntJob.Server/Setting.cs

@@ -5,7 +5,7 @@ namespace AntJob.Server;
 
 /// <summary>配置</summary>
 [Config("AntJob")]
-public class Setting : Config<Setting>
+public class AntJobSetting : Config<AntJobSetting>
 {
     #region 属性
     /// <summary>调试开关。默认true</summary>

AntJob.Server/Worker.cs

@@ -29,7 +29,7 @@ public class Worker : IHostedService
 
     public async Task StartAsync(CancellationToken cancellationToken)
     {
-        var set = Setting.Current;
+        var set = AntJobSetting.Current;
 
         // 实例化RPC服务端，指定端口，指定ServiceProvider，用于依赖注入获取接口服务层
         var server = new ApiServer(set.Port)

AntJob.Web/AntJob.Web.csproj

@@ -14,9 +14,7 @@
     <Deterministic>false</Deterministic>
     <OutputPath>..\Bin\Web</OutputPath>
     <AppendTargetFrameworkToOutputPath>false</AppendTargetFrameworkToOutputPath>
-    <DebugType>pdbonly</DebugType>
-    <Optimize>true</Optimize>
-    <DefineConstants>TRACE</DefineConstants>
+    <ImplicitUsings>enable</ImplicitUsings>
     <LangVersion>latest</LangVersion>
     <IsPackable>true</IsPackable>
     <ApplicationIcon>favicon.ico</ApplicationIcon>
@@ -36,6 +34,11 @@
     <Content Remove="Areas\Ant\Views\AppOnline\_List_Data.cshtml" />
     <Content Remove="Areas\Ant\Views\JobError\_List_Data.cshtml" />
   </ItemGroup>
+  <ItemGroup>
+    <Compile Include="..\AntJob.Server\Services\AppService.cs" Link="Services\AppService.cs" />
+    <Compile Include="..\AntJob.Server\Services\JobService.cs" Link="Services\JobService.cs" />
+    <Compile Include="..\AntJob.Server\Setting.cs" Link="Setting.cs" />
+  </ItemGroup>
   <ItemGroup>
     <Content Include="favicon.ico" />
   </ItemGroup>
@@ -49,5 +52,7 @@
   </ItemGroup>
   <ItemGroup>
     <Folder Include="Areas\Ant\Views\AppOnline\" />
+    <Folder Include="Controllers\" />
+    <Folder Include="Services\" />
   </ItemGroup>
 </Project>

AntJob.Web/Startup.cs

@@ -1,4 +1,5 @@
-using Microsoft.AspNetCore.Builder;
+using AntJob.Server.Services;
+using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -25,6 +26,9 @@ public class Startup
             set.Save();
         }
 
+        services.AddSingleton<AppService>();
+        services.AddSingleton<JobService>();
+
         services.AddControllersWithViews();
         services.AddCube();
     }