Ruby并发Rake任务中如何避免数据库竞争条件？

核心问题分析 #

你当前的代码先查询333条未处理记录，再逐个更新状态，这中间其他副本已经能查到这些记录并抢占，导致竞争。必须把查询未处理记录+标记为正在处理做成原子操作，彻底避免中间被其他实例插入。

最优解决方案（支持SKIP LOCKED的数据库） #

如果你的数据库支持SKIP LOCKED（如PostgreSQL 9.5+、MySQL 8.0+），这是最高效的方案，能直接跳过已被锁定的记录，原子性分配一批记录给当前实例：

# 原子性获取并锁定一批记录
processed_ids = MyModel.transaction do
  # 按主键排序查询未处理记录，锁定并跳过已被其他实例锁定的
  target_ids = MyModel.where.not(status: [:being_processed, :processed])
                     .order(:id) # 强制统一排序，彻底避免死锁
                     .limit(333)
                     .lock('SKIP LOCKED')
                     .pluck(:id)
  
  # 批量更新状态为being_processed
  MyModel.where(id: target_ids).update_all(status: :being_processed)
  
  target_ids
end

# 处理已锁定的记录
processed_ids.each do |item_id|
  item = MyModel.find(item_id)
  begin
    res = send_api_request(item.data1)
    item.update!(data2: res["data2"], status: :processed)
  rescue StandardError => e
    # 处理失败时重置状态，避免记录永久卡住
    item.update!(status: :failed, error_message: e.message)
  end
end

方案优势 #

• 原子性：整个查询+标记操作在事务内完成，其他副本无法抢占这批记录。
• 无等待：SKIP LOCKED让当前实例直接跳过已被锁定的记录，不用等待锁释放，效率更高。
• 低死锁风险：按主键排序查询，所有实例都按同一顺序锁定记录，彻底避免循环等待导致的死锁。

兼容老版本数据库的方案（无SKIP LOCKED支持） #

如果数据库不支持SKIP LOCKED，可以采用逐个原子锁定的方式，同样通过排序避免死锁：

loop do
  # 事务内查询并锁定单条记录
  item = MyModel.transaction do
    candidate = MyModel.where.not(status: [:being_processed, :processed])
                       .order(:id) # 统一排序，避免死锁
                       .limit(1)
                       .lock(true) # 等价于SELECT ... FOR UPDATE
                       .first
    if candidate
      candidate.update!(status: :being_processed)
      candidate
    else
      nil
    end
  end

  break unless item # 无可用记录时退出循环

  begin
    res = send_api_request(item.data1)
    item.update!(data2: res["data2"], status: :processed)
  rescue StandardError => e
    item.update!(status: :failed, error_message: e.message)
  end
end

注意事项 #

• 虽然是逐个处理，但按主键排序确保所有实例锁定记录的顺序一致，不会出现死锁。
• 批量处理的效率略低，但胜在兼容性好。

SELECT ... FOR UPDATE的适用性及死锁规避 #

SELECT ... FOR UPDATE完全适用，但需要注意两点：

1. 必须在事务内使用：单独的SELECT FOR UPDATE不会持久化锁，只有包裹在事务中才能保证原子性。
2. 强制统一排序：死锁的核心原因是不同实例按不同顺序锁定记录，只要所有查询都按同一字段（如主键）排序，就能避免循环等待，彻底消除死锁风险。

额外保障措施 #

• 超时记录清理：如果脚本崩溃，可能有记录停留在being_processed状态，需要定时任务（如Sidekiq Cron）重置超时记录（例如超过10分钟仍未处理完成的，将状态改回未处理）。
• 终态标记：处理完成后务必将状态更新为processed或failed，避免后续实例重复处理。

内容的提问来源于stack exchange，提问作者Koklushkin