长运行进程管理——Huginn 的 AgentRunner 框架

长运行进程管理——Huginn 的 AgentRunner 框架

学习目标

• 理解长运行 Agent 与定时 Agent 的区别
• 掌握 AgentRunner 的 Worker 生命周期管理
• 学会实现死亡自动重启与信号处理
• 了解互斥锁保护并发安全

项目实践

两类 Agent 的执行模型

AgentRunner 架构

Worker 生命周期

# 启动新 Worker
(new_worker_ids - current_worker_ids).each do |new_worker_id|
  @workers[new_worker_id] = workers[new_worker_id]
  @workers[new_worker_id].setup!(@scheduler, @mutex)
  @workers[new_worker_id].run!
end

# 停止已删除的 Worker
(current_worker_ids - new_worker_ids).each do |outdated_worker_id|
  @workers[outdated_worker_id].stop!
  @workers.delete(outdated_worker_id)
end

# 自动重启死亡的 Worker
def restart_dead_workers
  @workers.each_pair do |id, worker|
    if !worker.restarting && worker.thread && !worker.thread.alive?
      @workers[id].run!  # 重新启动
    end
  end
end

互斥锁保护

def with_mutex
  mutex.synchronize do
    ActiveRecord::Base.connection_pool.with_connection do
      yield
    end
  end
end

所有需要并发安全的操作（事件传播、过期清理、调度执行）都通过 with_mutex 保护。

两层锁：

1. @mutex（AgentRunner 级别）：保护 Worker 列表的变更
2. with_mutex（Scheduler 级别）：保护数据库操作的原子性

信号处理

def set_traps
  %w(INT TERM QUIT).each do |signal|
    Signal.trap(signal) { @signal_queue << signal }
  end
end

def handle_signal(signal)
  case signal
  when 'INT', 'TERM', 'QUIT'
    stop  # 停止所有 Worker 和 Scheduler
  end
end

非阻塞信号处理：信号不直接处理，而是推入队列，在主循环中消费——避免在信号处理器中执行复杂逻辑。

问题与规避

线程安全

Worker 在独立线程中运行，但 ActiveRecord 连接不是线程安全的。

Huginn 的处理：每个操作通过 ActiveRecord::Base.connection_pool.with_connection 获取独立连接。

Worker 启动风暴

当大量 Worker 同时启动时，可能耗尽数据库连接。

规避方式：AgentRunner 的 only 和 except 选项允许选择性启动 Worker。

连接泄漏

Worker 异常退出时可能未释放数据库连接。

Huginn 的处理：connection_pool.with_connection 块确保即使异常也会释放连接。

设计取舍

单进程多线程 vs 多进程

维度	单进程多 Worker（Huginn 默认）	多进程
内存占用	低（共享进程内存）	高（每个进程独立内存）
隔离性	低（一个 Worker 崩溃可能影响其他）	高
部署	简单	复杂

Huginn 也支持多进程模式（huginn-single-process Docker 镜像），在大规模部署时使用。

参考来源

• Ruby Threading — Ruby 线程文档
• LongRunnable — Ruby 长运行进程框架