Sidekiq中的作业重试次数

Question

有没有办法获取当前作业的重试计数？

我希望作业在x次重试后停止而不是崩溃。我想在执行方法中询问重试计数，这样如果重试计数等于x，我就可以简单地返回。

def perform(args)
  return if retry_count > 5
  ...
end

使用 Sidekiq 2.12。

编辑

我（不是 OP）有同样的问题，但出于不同的原因。如果任务正在重试，我想要进行额外的健全性检查，以确保仍需要该任务，并在自从它被排队以来发生了某些外部变化导致它不再可能成功时停止重试。

那么，有没有办法获取当前任务的重试计数？目前的答案只建议你绕开需要它的方式或者可以从作业外部获得它。

- Cimm

---

你有找到解决方案吗？ - RAJ

Answer 1

可以通过添加Sidekiq中间件来将msg['retry_count']设置为作业类的实例变量来实现。

添加一个中间件（在Rails中，通常是/config/initializers/文件夹中的文件）如下：

class SidekiqMiddleware
    def call(worker, job, queue)
        worker.retry_count = job['retry_count'] if worker.respond_to?(:retry_count=)
        yield
    end
end

Sidekiq.configure_server do |config|
    config.server_middleware do |chain|
        chain.add SidekiqMiddleware
    end
end

在你的工作中：

include Sidekiq::Worker
attr_accessor :retry_count

def retry_count
  @retry_count || 0
end

def perform(args)
  return if retry_count > 5
  ...
end

- Derek

---

我应该把中间件类放在哪个文件夹里？我可以把它放在我放“Sidekiq.configure_server”的地方吗？ - Henley

为什么我会收到这个错误 NoMethodError: undefined method retry_count=' for #Sidekiq::Extensions::DelayedMailer:0x007f9ad8928338`？ - Max Rose-Collins

实际上，虽然这种方法是有效的，但 retry_count 读取器的代码不正确，因为它将连续返回 0、0、1、2、3、4、5、6、7、8、9、10。双重 0 的原因是，在中间件级别上，第一次执行时不存在 :retry_count 键，当在第一次重试（作业的第二次执行）时添加键时，其值为 0（而不是此代码所假定的 1）。 - Dorian

它返回 nil、0、1、2、3、4 等。所以 if msg['retry_count'].nil? then retry_count = 0 else retry_count = msg['retry_count'] + 1 end - Vikrant Chaudhary

这个答案最初检查了 respond_to?(:retry_count)，但我改成了 respond_to?(:retry_count=)，因为它实际上使用的是这个方法。这可能是 @MaxRose-Collins 遇到 NoMethodError 的原因之一。 - Brian Underwood

Answer 2

你不需要直接处理这个逻辑来完成你想要的。只需像这样向你的worker添加一些配置.. 注意下面的sidekiq_options。根据你下面的评论，"防止Sidekiq将作业移动到死作业队列"

 class MyWorker  
     include Sidekiq::Worker
     sidekiq_options :retry => 5, :dead => false

      def perform
          #do some stuff
      end
 end

如果任务失败，它应该重试5次并优雅地失败。同时，如果您想在5次重试后执行一段代码块，worker有一个名为sidekiq_retries_exhausted的方法，您可以在其中进行自定义日志记录等操作。

- blotto

---

谢谢，但是这将在5次尝试后以失败的工作结束，我想尝试5次，如果不行，就停止而不会引发错误。那不一样。sidekiq_retries_exhausted也是同样的事情，已经太晚了，工作失败了。我想在它失败之前停止它。 - Cimm

也许我可以使用 sidekiq_retries_exhausted 来防止 Sidekiq 将作业移动到死亡作业队列？ - Cimm

哦，原来在Sidekiq 3.0中这真的很简单，回答已更新。 - blotto

好的，谢谢你。下周将升级Sidekiq并回报！ - Cimm

文档链接：https://github.com/mperham/sidekiq/wiki/Error-Handling - Constantin De La Roche

Answer 3

我的使用场景是为了防止在部署期间发生异常或停机而安排多个作业。因此，我需要重试计数。以上解决方案不适用于sidekiq~> 5.0.4，这是我经过测试的解决方案。

# config/initializers/sidekiq.rb

# define your middleware
module Sidekiq::Middleware::Server
  class SetRetryCountMiddleware
    def call(worker, job_params, _queue)
      retry_count = job_params["retry_count"]
      worker.instance_variable_set(:@retry_count, retry_count)
      yield
    end
  end
end

# add your defined middleware
Sidekiq.configure_server do |config|
  config.server_middleware do |chain|
    chain.add Sidekiq::Middleware::Server::SetRetryCountMiddleware
  end
  config.redis = {url: "redis://sidekiq:6379/0"}
  config.logger.level = Logger::INFO
end

在你的worker中使用&

class YetAnotherWorker < Base
  sidekiq_options  :queue => :critical, :retry => true

  def perform(args)
    begin
      # lines that might result in exception
    rescue => exception
      logger.warn("#{exception.class}")
      raise(exception)
    ensure
      # below line will ensure job is scheduled only once, avoiding multiple jobs if above lines throws an error
      schedule_next_run({my_key: "my_value"})
    end
  end

  def schedule_next_run(args)
    YetAnotherWorker.perform_at(Time.now + 7.days, args) if first_run
  end

  def first_run
    @retry_count.nil?
  end

end

此外，在第一次运行时，job_params 中没有可用的 retry_count 键，因此计数将显示为 nil,0,1,2..。

- Lubdhak

---

如果你不喜欢重试次数偏差1或以nil开头，你可以从我的答案中借用修正因子： (count.nil? ? 0 : 1).then { |correction_factor| count.to_i + correction_factor } - SMAG

Answer 4

您可以使用 Sidekiq API 访问重试次数：

https://github.com/mperham/sidekiq/wiki/API#retries

找到您所需的工作并使用job ['retry_count']来获取重试次数。

- aledalgrande

---

谢谢，但是在作业运行时从内部访问重试不再RetrySet中，因为它是活动的正在运行的作业。 - Cimm

Answer 5

我相信这可以在没有中间件的情况下完成，但不确定这种方法是否比中间件方法更有益。然而，以下是我所做的：

注意：此方法假设您正在使用Redis来排队作业。

我通过以下方式使我的作业能够访问Redis：

def redis
  # You may need to disable SSL, depending on your environment, if so use this to do so:
  # @redis = Redis.new(ssl_params: { verify_mode: OpenSSL::SSL::VERIFY_NONE })
  @redis = Redis.new
end

# You can do this directly on your job, in your BaseJob or in a module / concern, reader's choice on implementation.

def retry_count
  # retry_count is not present on the first attempt
  redis_work_json.dig("payload","retry_count").then do |count|
    # the correction factor will give us the retry counts that you would expect, since the actual counts lag by 1, as described in the other answers
    # NOTE: nil.to_i => 0
    (count.nil? ? 0 : 1).then { |correction_factor| count.to_i + correction_factor }
  end
end

辅助方法：

# convert the Redis data for the current job to JSON
def redis_work_json
  # we may be in a race condition with Redis to save the key we are looking for
  sleep(100) until redis.keys.any? { |key| key.include? ":work" }

  redis.keys.each do |key|
    next unless key.include? ":work"

    return nested_redis_value_with_jid(key).then do |value|
      next if value.nil?

      json_from(value)
    end
  end
end

# find the data stored in Redis for the current Job
def nested_redis_value_with_jid(key)
  # the work key will have a hash value so it needs to be fetched via Redis::Commands::Hashes
  # hvals will skip the random key that Redis nested this hash in
  # find the hash value that matches this job's jid
  redis.hvals(key).find { |v| v.include?(jid) }
end

def flatten_json_str(str)
  # This may seem gnarly but it allows `JSON.parse` to work at it's full potential
  # instead of manually using JSON.parse on nested JSON strings in a loop
  str.gsub(":\"{", ":{").gsub("}\",", "},").delete("\\")
end

def json_from(value)
  JSON.parse(flatten_json_str(value))
end

- SMAG