Jobs

• 1: Jobs 概述
• 2: 特性与概念
• 3: 操作指南：调度和处理触发的作业

1 - Jobs 概述

许多应用程序需要作业调度，或者需要在将来执行某个操作。Jobs API 是一个编排器，用于调度这些未来的作业，可以在特定时间或特定间隔执行。

Jobs API 不仅帮助你调度作业，而且在内部，Dapr 使用 Scheduler 服务来调度 actor reminders。

Dapr 中的 Jobs 由以下部分组成：

• Jobs API 构建块
• Scheduler 控制平面服务

查看示例场景。

工作原理

Jobs API 是一个作业调度器，而不是运行作业的执行器。该设计保证至少一次作业执行，优先考虑持久化和水平扩展而非精确性。这意味着：

• 保证： 作业永远不会在计划时间之前被调用。
• 不保证： 在到期时间之后调用作业的时间上限。

所有计划作业的作业详细信息和用户关联数据都存储在 Scheduler 服务中的嵌入式 Etcd 数据库中。

你可以使用 jobs 来：

• 延迟你的发布订阅消息。 你可以在未来的特定时间发布消息（例如：从今天起一周后，或特定的 UTC 日期/时间）。
• 调度应用程序之间的服务调用方法调用。

场景

作业调度在以下场景中可能很有用：

• 自动化数据库备份： 确保数据库每天备份以防止数据丢失。安排备份脚本每天凌晨 2 点运行，这将创建数据库备份并将其存储在安全位置。

• 定期数据处理和 ETL（提取、转换、加载）： 处理和转换来自各种来源的原始数据并将其加载到数据仓库中。安排 ETL 作业在特定时间运行（例如：每小时、每天）以获取新数据、处理它，并用最新信息更新数据仓库。

• 电子邮件通知和报告： 通过电子邮件接收每日销售报告和每周绩效摘要。安排一个作业生成所需的报告，并通过电子邮件发送，每天报告在每天早上 6 点发送，每周摘要在每周一早上 8 点发送。

• 维护任务和系统更新： 执行定期维护任务，如清除临时文件、更新软件和检查系统运行状况。安排各种维护脚本在非高峰时段运行，例如周末或深夜，以最大程度减少对用户的干扰。

• 金融交易的批处理： 处理大量需要批处理并在每个工作日结束时进行结算的交易。安排批处理作业在每个工作日下午 5 点运行，汇总当天的交易并执行必要的结算和对账。

Dapr 的 jobs API 确保这些场景中代表的任务始终如一且可靠地执行，无需手动干预，从而提高效率并降低错误风险。

功能

Jobs API 的主要功能允许你创建、检索和删除计划作业。默认情况下，当你创建一个名称已存在的作业时，操作会失败，除非你显式地将 overwrite 标志设置为 true。这确保现有作业不会被意外修改或覆盖。

跨多个副本调度作业

当你创建一个作业时，它不会替换同名的现有作业，除非你显式设置 overwrite 标志。这意味着每次创建作业时，它都会重置计数，并且只在该作业的嵌入式 etcd 中保留 1 条记录。因此，你不必担心创建和触发多个作业——只有最新的作业被记录和执行，即使你的所有应用程序在启动时都调度相同的作业。

Scheduler 服务使作业的调度能够跨多个副本扩展，同时保证作业仅由 1 个 Scheduler 服务实例触发。

试用 jobs API

你可以在你的应用程序中试用 jobs API。在安装 Dapr后，你就可以开始使用 jobs API，从操作指南：调度作业指南开始。

后续步骤

• 了解如何使用 jobs API
• 了解有关 Scheduler 控制平面服务的更多信息
• Jobs API 参考

2 - 特性与概念

现在你已经大致了解了 jobs 构建块 ，让我们深入了解 Dapr Jobs 和各种 SDK 包含的特性和概念。Dapr Jobs：

• 提供了一个强大且可扩展的 API，用于调度未来要触发的操作。
• 暴露了在所有支持的语言中都通用的多项功能。
• 使用持续时间值时支持亚秒级精度（例如 500ms）。实际触发精度可能因运行时而异；基于 Cron 的调度仅支持秒级精度。

作业标识

所有作业都使用区分大小写的作业名称进行注册。这些名称在整个与 Dapr 运行时交互的服务中应该是唯一的。名称用作创建和修改作业时的标识符，也用于指示触发的调用与哪个作业相关联。

任意时刻一个名称只能关联一个作业。默认情况下，尝试创建与现有作业同名的作业会报错。然而，如果 overwrite 标志设置为 true，则新作业会覆盖同名作业。

调度作业

可以通过以下方式调度作业：

• 使用 Cron 表达式、持续时间值或周期表达式的间隔
• 特定的日期和时间

对于所有基于时间的调度，如果通过 RFC3339 规范提供了带时区的时间戳，则使用该时区。如果未提供，则使用运行 Dapr 的服务器所在的时区。换句话说，除非在调度作业时另有指定，否则不要假设时间以 UTC 时区运行。

使用 Cron 表达式调度

使用 Cron 表达式调度作业按特定间隔执行时，表达式使用 6 个字段，字段值范围见下表：

示例 1

"0 30 * * * *" 每小时的半点触发。

示例 2

"0 15 3 * * *" 每天 03:15 触发。

使用持续时间值调度

可以使用 Go 持续时间字符串 调度作业，其中字符串由一个（可能有符号的）十进制数字序列组成，每个数字可带有可选的小数部分和单位后缀。有效的时间单位为 "ns"、"us"、"ms"、"s"、"m" 或 "h"。

示例 1

"2h45m" 每 2 小时 45 分钟触发一次。

示例 2

"37m25s" 每 37 分钟 25 秒触发一次。

使用周期表达式调度

支持以下周期表达式。"@every" 表达式也接受 Go 持续时间字符串。

使用特定日期/时间调度

也可以通过使用 RFC3339 规范 提供日期，将作业调度到特定时间点执行。

示例 1

"2025-12-09T16:09:53+00:00" 表示作业应在 2025 年 12 月 9 日 UTC 时间下午 4:09:53 执行。

定时触发器

当预定的 Dapr 作业被触发时，运行时根据服务启动时向 Dapr 注册的方式，通过 HTTP 或 gRPC 向调度该作业的服务发送消息。

gRPC

当作业到达预定的触发时间时，触发的作业通过以下回调函数发送回应用程序：

注意： 以下示例使用 Go 语言，但适用于任何支持 gRPC 的编程语言。

import rtv1 "github.com/dapr/dapr/pkg/proto/runtime/v1"
...
func (s *JobService) OnJobEventAlpha1(ctx context.Context, in *rtv1.JobEventRequest) (*rtv1.JobEventResponse, error) {
    // Handle the triggered job
}

此函数在 gRPC 服务器的上下文中处理触发的作业。设置服务器时，请确保注册回调服务器，该服务器在作业被触发时调用此函数：

...
js := &JobService{}
rtv1.RegisterAppCallbackAlphaServer(server, js)

在此设置下，你可以完全控制如何接收和处理触发的作业，因为它们通过此 gRPC 方法直接路由。

HTTP

如果应用程序启动时未向 Dapr 注册 gRPC 服务器，则 Dapr 会通过向端点 /job/<job-name> 发送 POST 请求来触发作业。请求体包含以下作业信息：

• Schedule：作业触发的时间
• RepeatCount：可选值，指示作业应重复的次数
• DueTime：可选的时间点，表示作业应执行的单次时间（如果非重复），或者调度生效的最早开始时间
• Ttl：可选值，指示作业何时过期
• Payload：包含作业调度时原始存储数据的字节集合
• Overwrite：标志，允许请求的作业覆盖已存在的同名作业
• FailurePolicy：作业的可选失败策略

DueTime 和 Ttl 字段将反映 RFC3339 时间戳值，反映作业最初调度时提供的时区。如果未提供时区，这些值表示运行 Dapr 的服务器使用的时区。

管理作业

虽然作业通过 API 调用创建，但你可以通过 dapr scheduler CLI 命令管理（列出、检查、删除、备份和恢复）作业。

列出作业

dapr scheduler list --filter app
NAME                           BEGIN     COUNT  LAST TRIGGER
app/my-app/my-job              -3.89s    1      2025-10-03T16:58:55Z
app/my-app/another-job         -3.89s    1      2025-10-03T16:58:55Z

dapr scheduler list -o wide
NAMESPACE  NAME                     BEGIN                 EXPIRATION   SCHEDULE     DUE TIME                   TTL   REPEATS  COUNT  LAST TRIGGER
default    app/my-app/my-job        2025-10-03T16:58:55Z               @every 5s   2025-10-03T17:58:55+01:00        100      1      2025-10-03T16:58:55Z

dapr scheduler get app/my-app/my-job -o yaml

删除作业

删除指定作业：

dapr scheduler delete app/my-app/my-job

删除应用的所有作业：

dapr scheduler delete-all app/my-app

备份和恢复作业

导出所有作业：

dapr scheduler export -o jobs-backup.bin

之后导入：

dapr scheduler import -f jobs-backup.bin

摘要

• 使用 Jobs API 从应用程序创建或更新作业。
• 使用 dapr scheduler CLI 查看、检查、备份或删除作业。
• 作业存储在 Dapr Scheduler 中，确保在重启和部署时的可靠性。

3 - 操作指南：调度和处理触发的作业

既然你已经了解了作业构建块提供了什么，让我们来看一个如何使用该 API 的示例。下面的代码示例描述了一个为数据库备份应用程序调度作业并在触发时间处理它们的应用程序，触发时间也称为作业因其到达 dueTime 而被发送回应用程序的时间。

启动 Scheduler 服务

当你在自托管模式或 Kubernetes 上运行 dapr init时，Dapr Scheduler 服务会自动启动。

设置 Jobs API

在你的代码中，设置和调度应用程序内的作业。

• .NET
• Go

//Define the types that we'll represent the job data with
internal sealed record BackupJobData([property: JsonPropertyName("task")] string Task, [property: JsonPropertyName("metadata")] BackupMetadata Metadata);
internal sealed record BackupMetadata([property: JsonPropertyName("DBName")]string DatabaseName, [property: JsonPropertyName("BackupLocation")] string BackupLocation);

using System.Text;
using System.Text.Json;
using Dapr.Jobs;
using Dapr.Jobs.Extensions;
using Dapr.Jobs.Models;
using Dapr.Jobs.Models.Responses;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddDaprJobsClient();
var app = builder.Build();

//Registers an endpoint to receive and process triggered jobs
var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(5));
app.MapDaprScheduledJobHandler((string jobName, ReadOnlyMemory<byte> jobPayload, ILogger logger, CancellationToken cancellationToken) => {
  logger?.LogInformation("Received trigger invocation for job '{jobName}'", jobName);
  switch (jobName)
  {
    case "prod-db-backup":
      // Deserialize the job payload metadata
      var jobData = JsonSerializer.Deserialize<BackupJobData>(jobPayload);
      
      // Process the backup operation - we assume this is implemented elsewhere in your code
      await BackupDatabaseAsync(jobData, cancellationToken);
      break;
  }
}, cancellationTokenSource.Token);

await app.RunAsync();

//Create a scope so we can access the registered DaprJobsClient
await using scope = app.Services.CreateAsyncScope();
var daprJobsClient = scope.ServiceProvider.GetRequiredService<DaprJobsClient>();

//Create the payload we wish to present alongside our future job triggers
var jobData = new BackupJobData("db-backup", new BackupMetadata("my-prod-db", "/backup-dir")); 

//Serialize our payload to UTF-8 bytes
var serializedJobData = JsonSerializer.SerializeToUtf8Bytes(jobData);

//Schedule our backup job to run every minute, but only repeat 10 times
await daprJobsClient.ScheduleJobAsync("prod-db-backup", DaprJobSchedule.FromDuration(TimeSpan.FromMinutes(1)),
    serializedJobData, repeats: 10);

package main

import (
    //...

	daprc "github.com/dapr/go-sdk/client"
	"github.com/dapr/go-sdk/examples/dist-scheduler/api"
	"github.com/dapr/go-sdk/service/common"
	daprs "github.com/dapr/go-sdk/service/grpc"
)

func main() {
    // Initialize the server
	server, err := daprs.NewService(":50070")
    // ...

	if err = server.AddJobEventHandler("prod-db-backup", prodDBBackupHandler); err != nil {
		log.Fatalf("failed to register job event handler: %v", err)
	}

	log.Println("starting server")
	go func() {
		if err = server.Start(); err != nil {
			log.Fatalf("failed to start server: %v", err)
		}
	}()
    // ...

    // Set up backup location
	jobData, err := json.Marshal(&api.DBBackup{
		Task: "db-backup",
		Metadata: api.Metadata{
			DBName:         "my-prod-db",
			BackupLocation: "/backup-dir",
		},
	},
	)
	// ...
}

    // ...
    // Set up the job
	job := daprc.Job{
		Name:     "prod-db-backup",
		Schedule: "@every 1s",
		Repeats:  10,
		Data: &anypb.Any{
			Value: jobData,
		},
	}

...
if err = server.AddJobEventHandler("prod-db-backup", prodDBBackupHandler); err != nil {
    log.Fatalf("failed to register job event handler: %v", err)
}

// ...

// At job trigger time this function is called
func prodDBBackupHandler(ctx context.Context, job *common.JobEvent) error {
	var jobData common.Job
	if err := json.Unmarshal(job.Data, &jobData); err != nil {
		// ...
	}

	var jobPayload api.DBBackup
	if err := json.Unmarshal(job.Data, &jobPayload); err != nil {
		// ...
	}
	fmt.Printf("job %d received:\n type: %v \n typeurl: %v\n value: %v\n extracted payload: %v\n", jobCount, job.JobType, jobData.TypeURL, jobData.Value, jobPayload)
	jobCount++
	return nil
}

运行 Dapr 边车

一旦你在应用程序中设置了 Jobs API，在终端窗口中使用以下命令运行 Dapr 边车。

• Go

dapr run --app-id=distributed-scheduler \
                --metrics-port=9091 \
                --dapr-grpc-port 50001 \
                --app-port 50070 \
                --app-protocol grpc \
                --log-level debug \
                go run ./main.go

后续步骤

• 详细了解 Scheduler 控制平面服务
• Jobs API 参考