Oban Designer

安装

# mix.exs
{:oban, "~> 2.18"}

# config/config.exs
config :my_app, Oban,
  repo: MyApp.Repo,
  queues: [default: 10, mailers: 20, webhooks: 50, events: 5],
  plugins: [
    Oban.Plugins.Pruner,
    {Oban.Plugins.Cron, crontab: [
      {"0 2 * * *", MyApp.Workers.DailyCleanup},
      {"*/5 * * * *", MyApp.Workers.MetricsCollector}
    ]}
  ]

# 在 application.ex 的 children 中：
{Oban, Application.fetch_env!(:my_app, Oban)}

生成 Oban 迁移文件：

mix ecto.gen.migration add_oban_jobs_table

defmodule MyApp.Repo.Migrations.AddObanJobsTable do
  use Ecto.Migration
  def up, do: Oban.Migration.up(version: 12)
  def down, do: Oban.Migration.down(version: 1)
end

Worker 实现

基础 Worker

defmodule MyApp.Workers.SendEmail do
  use Oban.Worker,
    queue: :mailers,
    max_attempts: 5,
    priority: 1

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"to" => to, "template" => template} = args}) do
    case MyApp.Mailer.deliver(to, template, args) do
      {:ok, _} -> :ok
      {:error, :temporary} -> {:error, "temporary failure"}  # 将重试
      {:error, :permanent} -> {:cancel, "invalid address"}   # 不再重试
    end
  end
end

返回值

队列配置

参见 references/worker-patterns.md 获取常见 Worker 模式。

队列容量建议

队列优先级

同一队列内的任务按优先级执行（0 表示最高优先级），请谨慎使用：

%{user_id: user.id}
|> MyApp.Workers.SendEmail.new(priority: 0)  # 紧急任务
|> Oban.insert()

重试策略

默认退避策略

Oban 使用指数退避：时间为 attempt^4 + attempt 秒。

自定义退避策略

defmodule MyApp.Workers.WebhookDelivery do
  use Oban.Worker,
    queue: :webhooks,
    max_attempts: 10

  @impl Oban.Worker
  def backoff(%Oban.Job{attempt: attempt}) do
    # 带抖动的指数退避：2^attempt + 随机值(0..30)
    trunc(:math.pow(2, attempt)) + :rand.uniform(30)
  end

  @impl Oban.Worker
  def perform(%Oban.Job{args: args}) do
    # ...
  end
end

超时设置

use Oban.Worker, queue: :media

@impl Oban.Worker
def timeout(%Oban.Job{args: %{"size" => "large"}}), do: :timer.minutes(10)
def timeout(_job), do: :timer.minutes(2)

唯一性保障

防止重复任务提交：

defmodule MyApp.Workers.SyncAccount do
  use Oban.Worker,
    queue: :default,
    unique: [
      period: 300,               # 5 分钟
      states: [:available, :scheduled, :executing, :retryable],
      keys: [:account_id]        # 依据该参数键值去重
    ]
end

唯一性选项

替换已有任务

%{account_id: id}
|> MyApp.Workers.SyncAccount.new(
  replace: [:scheduled_at],    # 若存在重复，则更新 scheduled_at
  schedule_in: 60
)
|> Oban.insert()

Cron 调度

# config.exs
plugins: [
  {Oban.Plugins.Cron, crontab: [
    {"0 */6 * * *", MyApp.Workers.DigestEmail},
    {"0 2 * * *", MyApp.Workers.DailyCleanup},
    {"0 0 1 * *", MyApp.Workers.MonthlyReport},
    {"*/5 * * * *", MyApp.Workers.HealthCheck, args: %{service: "api"}},
  ]}
]

Cron 表达式格式：分钟 小时 日期 月份 星期几。

插入任务

# 立即执行
%{user_id: user.id, template: "welcome"}
|> MyApp.Workers.SendEmail.new()
|> Oban.insert()

# 延迟执行
%{report_id: id}
|> MyApp.Workers.GenerateReport.new(schedule_in: 3600)
|> Oban.insert()

# 指定时间执行
%{report_id: id}
|> MyApp.Workers.GenerateReport.new(scheduled_at: ~U[2024-01-01 00:00:00Z])
|> Oban.insert()

# 批量插入
changesets = Enum.map(users, fn user ->
  MyApp.Workers.SendEmail.new(%{user_id: user.id})
end)
Oban.insert_all(changesets)

# 在 Ecto.Multi 中使用
Ecto.Multi.new()
|> Ecto.Multi.insert(:user, changeset)
|> Oban.insert(:welcome_email, fn %{user: user} ->
  MyApp.Workers.SendEmail.new(%{user_id: user.id})
end)
|> Repo.transaction()

Oban Pro 功能

需持有 Oban Pro 许可证方可使用：

Batch（任务组）

# 批量处理条目，全部完成后触发回调
batch = MyApp.Workers.ProcessItem.new_batch(
  items |> Enum.map(&%{item_id: &1.id}),
  callback: {MyApp.Workers.BatchComplete, %{batch_name: "import"}}
)
Oban.insert_all(batch)

Workflow（任务依赖）

Oban.Pro.Workflow.new()
|> Oban.Pro.Workflow.add(:extract, MyApp.Workers.Extract.new(%{file: path}))
|> Oban.Pro.Workflow.add(:transform, MyApp.Workers.Transform.new(%{}), deps: [:extract])
|> Oban.Pro.Workflow.add(:load, MyApp.Workers.Load.new(%{}), deps: [:transform])
|> Oban.insert_all()

Chunk（聚合多个任务）

defmodule MyApp.Workers.BulkIndex do
  use Oban.Pro.Workers.Chunk,
    queue: :indexing,
    size: 100,            # 每次处理 100 个任务
    timeout: 30_000       # 或最多执行 30 秒

  @impl true
  def process(jobs) do
    items = Enum.map(jobs, & &1.args)
    SearchIndex.bulk_upsert(items)
    :ok
  end
end

测试

详见 references/testing-oban.md 获取详细测试模式。

环境配置

# config/test.exs
config :my_app, Oban,
  testing: :manual  # 或设为 :inline 实现同步执行

# test_helper.exs（若使用 :manual 模式）
Oban.Testing.start()

断言任务已入队

use Oban.Testing, repo: MyApp.Repo

test "enqueues welcome email on signup" do
  {:ok, user} = Accounts.register(%{email: "test@example.com"})

  assert_enqueued worker: MyApp.Workers.SendEmail,
    args: %{user_id: user.id, template: "welcome"},
    queue: :mailers
end

在测试中执行任务

test "processes email delivery" do
  {:ok, _} =
    perform_job(MyApp.Workers.SendEmail, %{
      "to" => "user@example.com",
      "template" => "welcome"
    })
end

监控

Telemetry 事件

# 在 application.ex 中挂载
:telemetry.attach_many("oban-logger", [
  [:oban, :job, :start],
  [:oban, :job, :stop],
  [:oban, :job, :exception]
], &MyApp.ObanTelemetry.handle_event/4, %{})

关键指标

• 任务执行耗时（p50、p95、p99）
• 队列深度（各队列待处理任务数）
• 各 Worker 错误率
• 各 Worker 重试率