XiHan Framework 处于预览阶段

查看获取方式
曦寒开发框架文档
取本页 Markdown

集群与管理 ​

这一页接着作业与计划讲两件事：一是运行时管理，用 IJobScheduler 登记、暂停、恢复、触发、取消作业，查运行历史，恢复改过的代码作业；二是多实例部署，引擎怎样轮询与抢占、IJobStore 共享与否的差别、三种领导选举、调度事件、配置与错误码。读完你能给管理界面提供完整的作业操作，并把调度部署到多个实例上而不重复运行。

为什么需要 ​

作业一旦上线，就有人要在运行时动它：临时跑一次、出问题时暂停、改一下计划、看最近几次为什么失败。多实例部署又带来另一组问题：三个副本各跑一遍夜间清理，或者谁都以为别人在跑；跑着作业的节点宕了，那次运行就此蒸发。

框架把运行时管理收进一个批量优先的接口 IJobScheduler；多实例的协调交给两个窄契约：共享存储靠原子抢占保证每条到期触发器只被一个节点拿到，不共享的存储靠领导租约保证只有一个实例按计划运行。引擎的轮询、心跳、抢占、维护都在抽象包里实现一份，换 Provider 不改行为。

运行时管理：IJobScheduler ​

IJobScheduler（实现是 JobSchedulerEngine）是单例，所有方法都是批量的：以集合为入参，逐项返回结果，单条写法是 JobSchedulerExtensions 里的扩展方法。

方法作用主要错误
ScheduleAsync(definitions)新增或整体替换作业与触发器，其他节点下一个周期看到SCHEDULING_NOTFOUND_002、SCHEDULING_VALIDATION_001
UnscheduleAsync(names)删除作业与触发器；进行中的运行照常结束，历史保留SCHEDULING_NOTFOUND_001
PauseAsync / ResumeAsync(names)暂停或恢复作业的全部触发器；暂停期间手动触发仍可用，恢复后按 misfire 策略处理SCHEDULING_NOTFOUND_001
PauseTriggersAsync / ResumeTriggersAsync(keys)按 TriggerKey(job, trigger) 暂停或恢复单个触发器SCHEDULING_NOTFOUND_003
TriggerAsync(requests)在本实例立即运行并等待结果，按顺序逐项；不受暂停和领导权限制SCHEDULING_NOTFOUND_001、SCHEDULING_CONFLICT_001、SCHEDULING_UNAVAILABLE_001
EnqueueAsync(requests)插入一个立即到期的一次性触发器（manual- 前缀），由集群任一节点运行，不等待；返回触发器名SCHEDULING_NOTFOUND_001
CancelRunsAsync(runIds)本实例的运行立即取消令牌，其他节点的在下一个轮询周期取消SCHEDULING_NOTFOUND_004
StatusAsync(name) / StatusesAsync()作业状态：是否暂停、各触发器的状态与下一次时刻、运行中数量、最近一次完成的结果—
RunsAsync(query) / CountRunsAsync(query)运行历史与同一组过滤条件下的总数SCHEDULING_VALIDATION_002（RunsAsync 分页越界）
RestoreAsync(names)把运行时改过的代码作业恢复为代码定义SCHEDULING_NOTFOUND_005

JobTriggerRequest(Name, Data) 能从 JobName 隐式转换，手动触发带的数据覆盖作业数据的同名键。TriggerAsync 在本实例同步运行并拿到结果，适合「立即执行」；EnqueueAsync 只写存储，由集群任一节点运行，适合「排队跑一次」。

最小示例 ​

我们为每个租户在运行时建一个同步作业。运行时作业引用的类型要事先用 AddJobType<TJob>() 登记：

csharp
builder.Services.AddJobType<TenantSyncJob>();
csharp
using System.Text.Json.Nodes;
using XiHan.Framework.Primitives;
using XiHan.Framework.Scheduling.Abstractions.Jobs;
using XiHan.Framework.Scheduling.Abstractions.Schedulers;
using XiHan.Framework.Scheduling.Abstractions.Schedules;

public sealed class TenantSyncJob : IJob
{
    public ValueTask<Result> RunAsync(JobRun run, InvocationContext context, CancellationToken cancellationToken)
        => ValueTask.FromResult(Result.Success());
}

public sealed class TenantSyncPlanner(IJobScheduler scheduler)
{
    public async ValueTask<Result> EnableAsync(string tenantId, CancellationToken cancellationToken)
    {
        // 作业名不合法会抛 ArgumentException：租户 ID 只能含小写字母、数字、点、下划线与连字符
        var definition = new JobDefinition($"tenant-{tenantId}.sync", JobSchedule.Every(TimeSpan.FromMinutes(10)), typeof(TenantSyncJob))
        {
            Data = new JsonObject { ["tenant"] = tenantId },
            Retry = JobRetryPolicy.Fixed(3, TimeSpan.FromSeconds(30)),
        };
        return await scheduler.ScheduleAsync(definition, cancellationToken);
    }

    public async ValueTask<Result> DisableAsync(string tenantId, CancellationToken cancellationToken)
        => await scheduler.UnscheduleAsync($"tenant-{tenantId}.sync", cancellationToken);
}

调用 EnableAsync("acme") 后，tenant-acme.sync 每十分钟运行一次，失败后每 30 秒重试、最多三次。类型未登记时 ScheduleAsync 返回 SCHEDULING_NOTFOUND_002。多节点部署时每个节点都要登记同一组类型：节点抢到类型未登记的触发器时把它改为 Error 并记录原因。

查状态用 StatusAsync(name)，返回 Result<Maybe<JobStatus>>，JobStatus 里有 NextRun、Paused、RunningCount、各触发器的 NextFireAt 与 State、最近一次完成的结果。查历史用 RunsAsync(query)：JobRunQuery 按作业、状态集合、计划时刻范围（From 含、To 不含）、触发器名、执行节点、有无错误码过滤，Skip 缺省 0，Limit 缺省 100、最大 1000（JobRunQuery.MaxLimit）；结果按计划时刻降序。CountRunsAsync 用同一组过滤项给出总数，不看分页。分页越界不在构造时抛出，RunsAsync 返回 SCHEDULING_VALIDATION_002（HTTP 400）：查询常由请求绑定而来，越界是调用方可以改正的输入。

代码作业的运行时修改与指纹 ​

管理界面常要改代码作业的计划。ScheduleAsync 替换一个代码作业时，来源记为 Overridden，并保留代码定义的指纹（类型键、数据、标志、说明、各触发器的名字、计划文本、数据、优先级、misfire 策略与说明的 SHA-256；重试策略与执行时限设置了才计入）。下次启动同步代码作业时：

  1. 存储里是 Overridden 且指纹与当前代码定义相同：代码没改，保留运行时的修改。
  2. 其余情况：按代码写入，来源回到 Code；覆盖了运行时修改的记一条警告日志。
  3. 暂停状态一律保留。

RestoreAsync 随时把作业恢复为代码定义，保留暂停状态与同名触发器的进度；作业没有代码定义（来源 Runtime）时返回 SCHEDULING_NOTFOUND_005。删除代码作业只到下次启动为止，启动时会按代码重新写入。JobTypeRegistry.Keys 列出已登记的类型键，供界面新建作业时选择。

集群：引擎与存储 ​

JobSchedulerEngine 由 Provider 特性装配，在应用进入服务状态后开始轮询，每个 PollInterval（缺省 1 秒）执行一个周期：

  1. 应用不在服务状态（启动中、排空中）时什么都不做，只释放持有的领导租约，另一个在服务的实例下一个周期就接手。
  2. 首个周期把代码登记的作业写入存储；失败时记日志，下个周期重试。
  3. 存储共享时写节点心跳。
  4. 续约领导租约，没有时尝试获取（资源 LeaseResource，存活 LeaseTimeToLive）。
  5. 领导者每 HousekeepingInterval（缺省 60 秒）维护一次：恢复失联节点，删除结束早于 HistoryRetention（缺省 7 天）的历史。
  6. 取消本节点上被请求取消的运行。
  7. 运行到期的本地触发器与本实例的本地重试。
  8. 抢占到期触发器并在后台运行：名额 = min(AcquireBatchSize, MaxConcurrentRuns − 本节点运行中数量)。

IJobStore.IsShared 决定集群唯一的作业怎样保证只跑一次：

存储IsShared只跑一次的机制
Scheduling.Databasetrue抢占：AcquireDueAsync 原子地把到期触发器改成 Acquired(nodeId)，每条只被一个节点拿到；TryBeginRunAsync(exclusive: true) 原子地占用作业的 ActiveRunId，实现全集群串行。节点按周期写心跳，超过 NodeTimeout 即失联
Scheduling.InMemoryfalse领导权：只有持有租约的实例按计划运行集群唯一的作业，其他实例把抢到的按计划触发器放回等待；入队、重试这类一次性触发器只有本进程看得到，不受领导权限制

触发器的状态：

状态含义
Waiting等待到期
Paused已暂停
Acquired已被某节点抢占；间隔计划在运行结束前保持这个状态
Blocked到期但作业不允许并发、上一次未结束
Complete没有下一次（一次性已运行、次数用尽、超过 EndAt）
Error无法运行（作业类型未登记），原因在 Error 字段

停止时引擎先停轮询，再等待进行中的运行结束；宿主的停止令牌触发时取消这些运行，最后释放领导租约。启动之前与开始停止之后，本实例的 TriggerAsync 返回 SCHEDULING_UNAVAILABLE_001（HTTP 503）、不起新的运行；EnqueueAsync 只写存储，不受影响。

数据库 Provider 经 persistence.relational@1 的仓储读写，每个存储操作是一次独立的管道调用与事务，不加入调用方的工作单元；需要建五张表（作业、触发器、运行、节点、租约），每个节点的 XiHan:Scheduling:NodeId 必须唯一，见 Scheduling.Database。

领导选举 ​

ILeaderElection.TryAcquireAsync(resource, timeToLive) 在资源空闲或租约已过期时授予 LeaderLease，被他人持有时返回 SCHEDULING_CONFLICT_002。LeaderLease 有 Holder、ExpiresAt、IsHeld，RenewAsync 延长一个存活时间（已过期或被接管时同样返回 SCHEDULING_CONFLICT_002），ReleaseAsync 立即放弃。租约不可重入。

实现持有者存储不可达时
InMemoryLeaderElection每次获取新生成的随机值进程内字典，只在本进程内互斥—
DatabaseLeaderElection节点 ID（XiHan:Scheduling:NodeId）加随机令牌与作业同库的一行，并发戳裁决SCHEDULING_UNAVAILABLE_002，持有状态不变
KubernetesLeaderElectionPod 名coordination.k8s.io/v1 Lease，resourceVersion 裁决并发SCHEDULING_UNAVAILABLE_002，下个周期重试

应用自己的单例工作也可以用它竞争领导权：

csharp
using XiHan.Framework.Scheduling.Abstractions.Jobs;
using XiHan.Framework.Scheduling.Abstractions.Leadership;

public sealed class IndexRebuilder(ILeaderElection election)
{
    public async ValueTask<bool> TryRebuildAsync(CancellationToken cancellationToken)
    {
        var acquired = await election.TryAcquireAsync(new JobName("search.rebuild"), TimeSpan.FromMinutes(5), cancellationToken);
        if (!acquired.TryGetValue(out var lease))
        {
            return false;   // 别的实例正在做，或选举存储不可达
        }

        await using (lease)
        {
            return true;    // 长于存活时间的工作要定期 RenewAsync，续约失败时停下
        }
    }
}

在 Kubernetes 上，示例应用 KubernetesApp 在 Pod 内装配 KubernetesSchedulingFeature，在本地退回 Full 组合自带的进程内选举；JobSchedulerEngine.IsLeader 与 NodeId 能放进端点展示当前谁是领导者：

csharp
using XiHan.Framework.Profiles.Full;
using XiHan.Framework.Scheduling.Kubernetes;

// 在 Pod 里（API Server 的环境变量存在）用 Lease 做领导选举；本地运行退回进程内选举
var inCluster = Environment.GetEnvironmentVariable("KUBERNETES_SERVICE_HOST") is not null;
builder.AddXiHan(xihan =>
{
    xihan.UseFullProfile();
    if (inCluster)
    {
        xihan.UseFeature<KubernetesSchedulingFeature>();
    }
});

Lease 名缺省是 xihan-scheduler，用 XiHan:Scheduling:Kubernetes:Lease 修改；Pod 的 ServiceAccount 需要对 coordination.k8s.io 下的 leases 有 get、update、patch、create 权限，xihan pack 能按应用清单生成这份 Role，见 Kubernetes。

事件 ​

装配了消息能力时，引擎经 IMessagePublisher 发布 CloudEvents 信封：Source 为 /scheduling，Subject 为运行 ID（作业级事件为作业名），Data 是 SchedulingEventData 的 JSON（作业、触发器名、运行 ID、触发来源、状态、各时刻、错误码与消息、Attempt、RetryOf、节点）。没有发布者时跳过，发布失败只记日志。

主题（SchedulingTopics）时机
scheduling.job.scheduled / scheduling.job.unscheduled作业登记或替换 / 删除
scheduling.job.paused / scheduling.job.resumed作业暂停 / 恢复
scheduling.run.started运行开始
scheduling.run.succeeded / scheduling.run.failed / scheduling.run.canceled运行结束
scheduling.run.lost运行所在节点失联

订阅方式见发布与订阅。Kafka 关了自动建主题时这 9 个主题要预先建好，见 Messaging.Kafka。

配置 ​

引擎选项在 XiHan:Scheduling（SchedulerOptions），两个 Provider 读同一节；值不合法时启动抛 ArgumentException。

键类型缺省说明
PollIntervalSeconds正数1轮询周期
NodeId字符串机器名:进程 ID:8 位随机节点身份：心跳、抢占、租约持有者都用它；多节点时每个节点唯一
MaxConcurrentRuns≥ 132本节点同时运行的上限
AcquireBatchSize≥ 116一个周期最多抢占的触发器数
MisfireThresholdSeconds≥ 060落后超过它算 misfire
LeaseResource作业名文法scheduling.jobs领导租约的资源名
LeaseTimeToLiveSeconds正数30领导租约的存活时间
NodeTimeoutSeconds正数90心跳超过它即失联
HousekeepingIntervalSeconds正数60领导者维护周期
HistoryRetentionDays≥ 07运行历史保留天数，0 为不删

各 Provider 自己的键（XiHan:Scheduling:InMemory:MaxRuns、XiHan:Scheduling:Database:HealthProbeIntervalSeconds、XiHan:Scheduling:Kubernetes:*）见对应包页。

错误码 ​

错误码含义
SCHEDULING_NOTFOUND_001作业未登记
SCHEDULING_NOTFOUND_002作业类型未登记
SCHEDULING_NOTFOUND_003触发器不存在
SCHEDULING_NOTFOUND_004运行不存在或已结束
SCHEDULING_NOTFOUND_005作业没有代码定义，不能恢复
SCHEDULING_VALIDATION_001作业定义不合法：触发器名重复、日历未登记
SCHEDULING_VALIDATION_002运行历史查询的分页越界（HTTP 400）
SCHEDULING_CONFLICT_001作业正在运行且不允许并发
SCHEDULING_CONFLICT_002领导租约被其他持有者占用
SCHEDULING_CONFLICT_003运行被取消
SCHEDULING_UNAVAILABLE_001本实例的调度器尚未启动或已开始停止，TriggerAsync 不起新的运行
SCHEDULING_UNAVAILABLE_002领导选举的存储（数据库、API Server）不可达或超时
SCHEDULING_UNAVAILABLE_003作业存储不可达
SCHEDULING_UNAVAILABLE_004作业存储容量已满（内存实现的运行历史全部在运行中）
SCHEDULING_TIMEOUT_001运行超过作业的执行时限
SCHEDULING_UNEXPECTED_001作业抛出未处理异常

错误文案在消息目录 scheduling 里，十种区域都有译文，见全球化和本地化。

常见问题 ​

三个副本各跑了一遍夜间作业 ​

只装配了 Scheduling.InMemory：每个副本有自己的存储和进程内租约，都当自己是领导者。在 Kubernetes 上加 KubernetesSchedulingFeature；需要运行时登记的作业在副本间可见、节点宕机后恢复运行，换 DatabaseSchedulingFeature。

领导者 Pod 滚动更新时会不会停一阵 ​

不会等租约到期。排空中的实例在每个周期释放租约，另一个在服务的实例下一个周期就接手。只有实例被强杀、没来得及释放时，才要等 LeaseTimeToLive（缺省 30 秒）过期。

TriggerAsync 返回 SCHEDULING_UNAVAILABLE_001 ​

本实例的引擎还没启动或已经开始停止，这两段时间没有轮询照看运行，所以不起新的。排空期间要排队运行的用 EnqueueAsync，它只写存储，由别的节点运行。

运行时改了作业计划，重新发布后被覆盖了 ​

代码定义变了。只有代码定义的指纹与存储里 Overridden 记录保存的指纹相同，运行时修改才保留；代码一改就以代码为准并记警告。要让修改长期有效，把它写回代码；临时调整用 PauseAsync，暂停状态一律保留。

相关包与下一步 ​

Released under The MIT License

Copyright ©2021-Present XiHanFun and contributors.