异步任务队列——Huginn 的 ActiveJob 多适配器架构

异步任务队列——Huginn 的 ActiveJob 多适配器架构

学习目标

• 理解 Huginn 的六类后台任务及其职责划分
• 掌握 ActiveJob 多适配器（Delayed Job / Resque）的适配策略
• 学会实现防重复入队检查机制
• 了解任务失败处理与自动清理策略

项目实践

六类后台任务

Job 类	触发方式	职责
AgentCheckJob	定时调度触发	加载 Agent 并调用 check 方法
AgentReceiveJob	传播器触发	加载 Event 并调用 Agent 的 receive 方法
AgentRunScheduleJob	调度器触发	调用 Agent.run_schedule(schedule_name)
AgentPropagateJob	每分钟定时器	调用 Agent.receive! 扫描新事件
AgentCleanupExpiredJob	每 6 小时定时器	清理 expires_at 过期的事件
AgentReemitJob	手动触发	重新发出特定事件

多适配器适配策略

Huginn 使用 ActiveJob 抽象层，支持多种后端：

Delayed Job（默认）：

• 任务存储在 delayed_jobs 数据库表中
• DelayedJobWorker 从表中轮询待执行任务
• 优势：无需额外基础设施，数据天然持久化
• 劣势：数据库读写开销

Resque（可选）：

• 任务存储在 Redis 队列中
• 需要额外部署 Redis 和 Resque Worker
• 优势：更高吞吐，更适合大规模部署

防重复入队检查

AgentPropagateJob 实现了防重复入队检查：

def self.can_enqueue?
  case queue_adapter.class.name
  when 'ActiveJob::QueueAdapters::DelayedJobAdapter'
    Delayed::Job.where(failed_at: nil, queue: 'propagation').count == 0
  when 'ActiveJob::QueueAdapters::ResqueAdapter'
    Resque.size('propagation') == 0 &&
    Resque.workers.select { |w| w.job && w.job['queue'] == 'propagation' }.count == 0
  else
    raise NotImplementedError, "unsupported adapter"
  end
end

每种适配器有独立的检查逻辑：

• Delayed Job：查询数据库表中是否有未失败的传播任务
• Resque：检查 Redis 队列大小和活跃 Worker 状态

任务失败处理

失败任务清理（每小时执行）：

def cleanup_failed_jobs!
  num_to_keep = ENV['FAILED_JOBS_TO_KEEP'].presence || 100
  first_to_delete = Delayed::Job
    .where.not(failed_at: nil)
    .order("failed_at DESC")
    .offset(num_to_keep)
    .limit(1)
    .pluck(:failed_at).first
  Delayed::Job.where(["failed_at <= ?", first_to_delete]).delete_all if first_to_delete
end

问题与规避

任务序列化

ActiveJob 需要序列化 Job 参数。传递 ActiveRecord 对象会导致全对象序列化，影响性能。

Huginn 的处理：传递 ID 而非对象（如 Agent.async_receive(agent_id, event_ids)），在 perform 方法中重新加载。

队列饥饿

当大量 CheckJob 占据队列时，PropagateJob 可能被延迟。

规避方式：使用不同的队列名（queue_as :propagation），配合多 Worker 优先级处理。

数据库连接池

大量并发 Job 可能耗尽数据库连接池。

Huginn 的配置：DATABASE_POOL=20（默认），根据 Worker 数量调整。

设计取舍

内建队列 vs 外部队列服务

维度	Delayed Job（数据库）	Sidekiq（Redis）
部署复杂度	低（无额外组件）	中（需要 Redis）
可靠性	高（数据库事务）	中（Redis 持久化可选）
吞吐量	低（数据库写入）	高（内存操作）
监控	需自建	成熟生态

Huginn 的默认选择降低了部署门槛——个人用户不需要部署 Redis 即可运行。

参考来源

• Active Job Basics — Rails Active Job 文档
• Delayed Job — 数据库后台任务库