如何通过cron job配合Sidekiq处理Redis队列的入队任务

不需要额外引入其他组件，直接基于现有栈就能实现需求，核心逻辑是把待处理webhook放到专属隔离队列，不被普通Sidekiq进程消费，再通过定时任务按频率批量拉取处理。

1. Webhook接收端仅做入队操作 #

Twilio webhook接口收到请求后，不执行任何业务逻辑，只做必要的签名校验，校验通过后直接把请求原始参数推入专门的待处理队列，立刻给Twilio返回200响应，避免Twilio因超时重复推送。

# app/controllers/twilio_webhooks_controller.rb
class TwilioWebhooksController < ApplicationController
  skip_before_action :verify_authenticity_token
  before_action :verify_twilio_request_signature

  def create
    raw_payload = request.request_parameters.to_h
    # 推入专属待处理队列，不和其他业务任务混用
    Sidekiq::Client.push(
      queue: 'twilio_pending_webhooks',
      class: 'TwilioWebhookDummyJob', # 空Job即可，不需要写perform逻辑，仅作为队列载体
      args: [raw_payload],
      retry: 3
    )
    head :ok
  end

  private
  def verify_twilio_request_signature
    # 保留原有Twilio签名校验逻辑，拦截伪造请求
  end
end

注意：这个twilio_pending_webhooks队列不要加入Sidekiq的默认监听列表，否则任务入队后会被立刻消费，达不到攒批按频率处理的效果。

2. 配置定时批处理任务 #

用sidekiq-cron做定时调度最适配现有Sidekiq栈，会自动加分布式锁避免多实例部署时重复执行，不需要额外维护系统级crontab。
首先添加依赖后，在初始化文件里配置定时任务，调整cron表达式即可直接控制整体处理频次：

# config/initializers/sidekiq_cron.rb
Sidekiq.configure_server do |config|
  config.on(:startup) do
    schedule_config = {
      twilio_webhook_batch_runner: {
        class: 'TwilioWebhookBatchProcessJob',
        # 修改这里的cron表达式即可调整处理频率，比如*/5 * * * * 是每5分钟执行一次
        cron: '*/5 * * * *',
        queue: 'critical' # 定时任务本身走高优先级队列，保证准点执行
      }
    }
    Sidekiq::Cron::Job.load_from_hash(schedule_config)
  end
end

批处理Job的核心逻辑是每次执行时从待处理队列拉取定量任务，匹配预设条件后再执行处理：

# app/jobs/twilio_webhook_batch_process_job.rb
class TwilioWebhookBatchProcessJob
  include Sidekiq::Job
  sidekiq_options queue: 'critical', retry: 2

  # 单次执行最大处理任务数，根据单条任务处理耗时调整，避免执行超时
  BATCH_LIMIT = 200

  def perform
    pending_queue = Sidekiq::Queue.new('twilio_pending_webhooks')
    processed = 0

    while processed < BATCH_LIMIT && !pending_queue.empty?
      job = pending_queue.first
      break if job.nil?
      job.delete # 从队列取出后先移除，避免重复消费

      begin
        payload = job.args.first
        # 匹配预设处理条件
        if satisfy_process_rules?(payload)
          run_webhook_business_logic(payload)
        else
          # 不符合处理条件的任务，按需选择丢弃、记日志或重新入队延后处理
          requeue_if_needed(payload)
        end
      rescue StandardError => e
        Rails.logger.error "[TwilioWebhook] Process failed: #{e.message}, payload: #{payload}"
        # 单条任务异常不影响整批执行，按需决定是否重新入队
      end

      processed += 1
    end
  end

  private
  def satisfy_process_rules?(payload)
    # 替换为实际的预设判断逻辑，比如消息类型、发送方、时间窗口校验等
    true
  end

  def requeue_if_needed(payload)
    # 替换为需要延后处理的重入队逻辑
  end

  def run_webhook_business_logic(payload)
    # 替换为实际的webhook业务处理代码
  end
end

3. 必要配置调整 #

修改Sidekiq的队列配置文件，确保待处理队列不被普通Worker监听：

# config/sidekiq.yml
:queues:
  - critical
  - default
  - mailers
  # 注意：不要把 twilio_pending_webhooks 加入监听列表

如果不想用sidekiq-cron，也可以写一个Rake任务封装上面的批处理逻辑，用系统级crontab调度，但是多实例部署时需要自己加分布式锁避免重复执行，维护成本更高。

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