UNIVPLMDataIntegration/Admin.NET/Admin.NET.Core/Service/Job/JobClusterServer.cs

// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
//
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
//
// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！

namespace Admin.NET.Core.Service;

/// <summary>
/// 作业集群控制
/// </summary>
public class JobClusterServer : IJobClusterServer
{
    private readonly IServiceScopeFactory _serviceScopeFactory;
    private readonly Random _random = new(DateTime.Now.Millisecond);

    public JobClusterServer(IServiceScopeFactory serviceScopeFactory)
    {
        _serviceScopeFactory = serviceScopeFactory;
    }

    /// <summary>
    /// 当前作业调度器启动通知
    /// </summary>
    /// <param name="context">作业集群服务上下文</param>
    public void Start(JobClusterContext context)
    {
        using var scope = _serviceScopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();

        // 在作业集群表中，如果 clusterId 不存在，则新增一条（否则更新一条），并设置 status 为 ClusterStatus.Waiting
        if (db.Queryable<SysJobCluster>().Any(u => u.ClusterId == context.ClusterId))
        {
            db.Updateable<SysJobCluster>().SetColumns(u => u.Status == ClusterStatus.Waiting).Where(u => u.ClusterId == context.ClusterId).ExecuteCommand();
        }
        else
        {
            db.Insertable(new SysJobCluster { ClusterId = context.ClusterId, Status = ClusterStatus.Waiting }).ExecuteCommand();
        }
    }

    /// <summary>
    /// 等待被唤醒
    /// </summary>
    /// <param name="context">作业集群服务上下文</param>
    /// <returns><see cref="Task"/></returns>
    public async Task WaitingForAsync(JobClusterContext context)
    {
        var clusterId = context.ClusterId;

        while (true)
        {
            // 控制集群心跳频率（放在头部为了防止 IsAnyAsync continue 没sleep占用大量IO和CPU）
            await Task.Delay(3000 + _random.Next(500, 1000)); // 错开集群同时启动

            try
            {
                ICache cache = App.GetRequiredService<ICacheProvider>().Cache;
                // 使用分布式锁
                using (cache.AcquireLock("lock:JobClusterServer:WaitingForAsync", 1000))
                {
                    // 在这里查询数据库，根据以下两种情况处理
                    // 1) 如果作业集群表已有 status 为 ClusterStatus.Working 则继续循环
                    // 2) 如果作业集群表中还没有其他服务或只有自己，则插入一条集群服务或调用 await WorkNowAsync(clusterId); 之后 return;
                    // 3) 如果作业集群表中没有 status 为 ClusterStatus.Working 的，调用 await WorkNowAsync(clusterId); 之后 return;
                    using var scope = _serviceScopeFactory.CreateScope();
                    var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
                    var sysJobClusterRep = App.GetRequiredService<SqlSugarRepository<SysJobCluster>>();
                    if (await db.Queryable<SysJobCluster>().AnyAsync(u => u.Status == ClusterStatus.Working)) continue;

                    await WorkNowAsync(clusterId);
                    return;
                }
            }
            catch
            {
            }
        }
    }

    /// <summary>
    /// 当前作业调度器停止通知
    /// </summary>
    /// <param name="context">作业集群服务上下文</param>
    public void Stop(JobClusterContext context)
    {
        // 在作业集群表中，更新 clusterId 的 status 为 ClusterStatus.Crashed
        using var scope = _serviceScopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
        db.Updateable(new SysJobCluster { Status = ClusterStatus.Crashed }).Where(u => u.ClusterId == context.ClusterId).ExecuteCommand();
    }

    /// <summary>
    /// 当前作业调度器宕机
    /// </summary>
    /// <param name="context">作业集群服务上下文</param>
    public void Crash(JobClusterContext context)
    {
        // 在作业集群表中，更新 clusterId 的 status 为 ClusterStatus.Crashed
        using var scope = _serviceScopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
        db.Updateable(new SysJobCluster { Status = ClusterStatus.Crashed }).Where(u => u.ClusterId == context.ClusterId).ExecuteCommand();
    }

    /// <summary>
    /// 指示集群可以工作
    /// </summary>
    /// <param name="clusterId">集群 Id</param>
    /// <returns></returns>
    private async Task WorkNowAsync(string clusterId)
    {
        // 在作业集群表中，更新 clusterId 的 status 为 ClusterStatus.Working
        using var scope = _serviceScopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
        await db.Updateable(new SysJobCluster { Status = ClusterStatus.Working }).Where(u => u.ClusterId == clusterId).ExecuteCommandAsync();
    }
}