namespace Admin.NET.Core.Service;
///
/// 作业集群控制
///
public class JobClusterServer : IJobClusterServer
{
private readonly SqlSugarRepository _sysJobClusterRep;
public JobClusterServer(SqlSugarRepository sysJobClusterRep)
{
_sysJobClusterRep = sysJobClusterRep;
}
///
/// 当前作业调度器启动通知
///
/// 作业集群服务上下文
public async void Start(JobClusterContext context)
{
// 在作业集群表中,如果 clusterId 不存在,则新增一条(否则更新一条),并设置 status 为 ClusterStatus.Waiting
if (await _sysJobClusterRep.IsAnyAsync(u => u.ClusterId == context.ClusterId))
{
await _sysJobClusterRep.UpdateSetColumnsTrueAsync(u => new SysJobCluster { Status = ClusterStatus.Waiting }, u => u.ClusterId == context.ClusterId);
}
else
{
await _sysJobClusterRep.InsertAsync(new SysJobCluster { ClusterId = context.ClusterId, Status = ClusterStatus.Waiting });
}
}
///
/// 等待被唤醒
///
/// 作业集群服务上下文
///
public async Task WaitingForAsync(JobClusterContext context)
{
var clusterId = context.ClusterId;
while (true)
{
try
{
// 在这里查询数据库,根据以下两种情况处理
// 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;
WorkNowAsync(clusterId);
return;
}
catch { }
// 控制集群心跳频率
await Task.Delay(3000);
}
}
///
/// 当前作业调度器停止通知
///
/// 作业集群服务上下文
public async void Stop(JobClusterContext context)
{
// 在作业集群表中,更新 clusterId 的 status 为 ClusterStatus.Crashed
await _sysJobClusterRep.UpdateSetColumnsTrueAsync(u => new SysJobCluster { Status = ClusterStatus.Crashed }, u => u.ClusterId == context.ClusterId);
}
///
/// 当前作业调度器宕机
///
/// 作业集群服务上下文
public async void Crash(JobClusterContext context)
{
// 在作业集群表中,更新 clusterId 的 status 为 ClusterStatus.Crashed
await _sysJobClusterRep.UpdateSetColumnsTrueAsync(u => new SysJobCluster { Status = ClusterStatus.Crashed }, u => u.ClusterId == context.ClusterId);
}
///
/// 指示集群可以工作
///
/// 集群 Id
///
private async void WorkNowAsync(string clusterId)
{
// 在作业集群表中,更新 clusterId 的 status 为 ClusterStatus.Working
await _sysJobClusterRep.UpdateSetColumnsTrueAsync(u => new SysJobCluster { Status = ClusterStatus.Working }, u => u.ClusterId == clusterId);
}
}