namespace Admin.NET.Core.Service; /// /// 作业集群控制 /// public class JobClusterServer : IJobClusterServer { private readonly Random rd = new(DateTime.Now.Millisecond); public JobClusterServer() { } /// /// 当前作业调度器启动通知 /// /// 作业集群服务上下文 public async void Start(JobClusterContext context) { var _sysJobClusterRep = App.GetService>(); // 在作业集群表中,如果 clusterId 不存在,则新增一条(否则更新一条),并设置 status 为 ClusterStatus.Waiting if (await _sysJobClusterRep.IsAnyAsync(u => u.ClusterId == context.ClusterId)) { await _sysJobClusterRep.AsUpdateable().SetColumns(u => u.Status == ClusterStatus.Waiting).Where(u => u.ClusterId == context.ClusterId).ExecuteCommandAsync(); } else { await _sysJobClusterRep.AsInsertable(new SysJobCluster { ClusterId = context.ClusterId, Status = ClusterStatus.Waiting }).ExecuteCommandAsync(); } } /// /// 等待被唤醒 /// /// 作业集群服务上下文 /// public async Task WaitingForAsync(JobClusterContext context) { var clusterId = context.ClusterId; while (true) { // 控制集群心跳频率(放在头部为了防止 IsAnyAsync continue 没sleep占用大量IO和CPU) await Task.Delay(3000 + rd.Next(500, 1000)); // 错开集群同时启动 try { ICache _cache = App.GetService(); //使用分布式锁 using (_cache.AcquireLock("lock:JobClusterServer:WaitingForAsync", 1000)) { var _sysJobClusterRep = App.GetService>(); // 在这里查询数据库,根据以下两种情况处理 // 1) 如果作业集群表已有 status 为 ClusterStatus.Working 则继续循环 // 2) 如果作业集群表中还没有其他服务或只有自己,则插入一条集群服务或调用 await WorkNowAsync(clusterId); 之后 return; // 3) 如果作业集群表中没有 status 为 ClusterStatus.Working 的,调用 await WorkNowAsync(clusterId); 之后 return; if (await _sysJobClusterRep.IsAnyAsync(u => u.Status == ClusterStatus.Working)) continue; await WorkNowAsync(clusterId); return; } } catch { } } } /// /// 当前作业调度器停止通知 /// /// 作业集群服务上下文 public async void Stop(JobClusterContext context) { var _sysJobClusterRep = App.GetService>(); // 在作业集群表中,更新 clusterId 的 status 为 ClusterStatus.Crashed await _sysJobClusterRep.UpdateSetColumnsTrueAsync(u => new SysJobCluster { Status = ClusterStatus.Crashed }, u => u.ClusterId == context.ClusterId); } /// /// 当前作业调度器宕机 /// /// 作业集群服务上下文 public async void Crash(JobClusterContext context) { var _sysJobClusterRep = App.GetService>(); // 在作业集群表中,更新 clusterId 的 status 为 ClusterStatus.Crashed await _sysJobClusterRep.UpdateSetColumnsTrueAsync(u => new SysJobCluster { Status = ClusterStatus.Crashed }, u => u.ClusterId == context.ClusterId); } /// /// 指示集群可以工作 /// /// 集群 Id /// private async Task WorkNowAsync(string clusterId) { var _sysJobClusterRep = App.GetService>(); // 在作业集群表中,更新 clusterId 的 status 为 ClusterStatus.Working await _sysJobClusterRep.UpdateSetColumnsTrueAsync(u => new SysJobCluster { Status = ClusterStatus.Working }, u => u.ClusterId == clusterId); } }