ruby-on-rails - 如何防止许多 sidekiq 作业超过 API 调用限制

Question

我正在开发一个 Ruby On Rails 应用程序。我们有许多 sidekiq worker 可以同时处理多个作业。每个作业都会调用 Shopify API，Shopify 设置的调用限制是 每秒 2 次调用。我想同步它，以便在给定的一秒钟内只有两个作业可以调用 API。我现在这样做的方式是这样的:

# frozen_string_literal: true
class Synchronizer

  attr_reader :shop_id, :queue_name, :limit, :wait_time

  def initialize(shop_id:, queue_name:, limit: nil, wait_time: 1)
    @shop_id = shop_id
    @queue_name = queue_name.to_s
    @limit = limit
    @wait_time = wait_time
  end

  # This method should be called for each api call
  def synchronize_api_call
    raise "a block is required." unless block_given?
    get_api_call
    time_to_wait = calculate_time_to_wait
    sleep(time_to_wait) unless Rails.env.test? || time_to_wait.zero?
    yield
  ensure
    return_api_call
  end

  def set_api_calls
    redis.del(api_calls_list)
    redis.rpush(api_calls_list, calls_list)
  end

  private

  def get_api_call
    logger.log_message(synchronizer: 'Waiting for api call', color: :yellow)
    @api_call_timestamp = redis.brpop(api_calls_list)[1].to_i
    logger.log_message(synchronizer: 'Got api call.', color: :yellow)
  end

  def return_api_call
    redis_timestamp = redis.time[0]
    redis.rpush(api_calls_list, redis_timestamp)
  ensure
    redis.ltrim(api_calls_list, 0, limit - 1)
  end

  def last_call_timestamp
    @api_call_timestamp
  end

  def calculate_time_to_wait
    current_time = redis.time[0]
    time_passed = current_time - last_call_timestamp.to_i
    time_to_wait = wait_time - time_passed
    time_to_wait > 0 ? time_to_wait : 0
  end

  def reset_api_calls
    redis.multi do |r|
      r.del(api_calls_list)
    end
  end

  def calls_list
    redis_timestamp = redis.time[0]
    limit.times.map do |i|
      redis_timestamp
    end
  end

  def api_calls_list
    @api_calls_list ||= "api-calls:shop:#{shop_id}:list"
  end

  def redis
    Thread.current[:redis] ||= Redis.new(db: $redis_db_number)
  end

end

我的使用方式是这样的

synchronizer = Synchronizer.new(shop_id: shop_id, queue_name: 'shopify_queue', limit: 2, wait_time: 1)
# this is called once the process started, i.e. it's not called by the jobs themselves but by the App from where the process is kicked off.
syncrhonizer.set_api_calls # this will populate the api_calls_list with 2 timestamps, those timestamps will be used to know when the last api call has been sent.

然后当一个工作要打电话的时候

syncrhonizer.synchronize_api_call do
   # make the call  
end

问题

问题在于，如果由于某种原因，作业无法将它调用的 api_call 返回到 api_calls_list，这将使该作业和其他作业永远停滞不前，或者直到我们注意到然后 我们再次调用 set_api_calls。这个问题不仅会影响那个特定的商店，还会影响其他商店，因为 sidekiq worker 是在使用我们应用程序的所有商店之间共享的。有时我们没有注意到，直到用户调用我们，我们发现它卡住了好几个小时，而它应该在几分钟内完成。

问题

我最近才意识到 Redis 并不是共享锁的最佳工具。所以我想问，还有其他好的工具可以胜任这项工作吗？？ 如果不是在 Ruby 世界，我也想向其他人学习。我对技术和工具都很感兴趣。所以每一点都有帮助。

Answer 1

您可能想要重组您的代码并创建一个微服务来处理 API 调用，这将使用本地锁定机制并强制您的工作人员在套接字上等待。它伴随着维护微服务的额外复杂性。但如果您赶时间，那么 Ent-Rate-Limiting 看起来也很酷。