Sidekiq: убедитесь, что все задания в очереди уникальны

У меня есть триггеры обновления, которые помещают задания в очередь Sidekiq. Поэтому в некоторых случаях для обработки одного и того же объекта может выполняться несколько заданий.

Существует несколько уникальных плагинов ( "Middleware" , Уникальные вакансии), они не документированы много, но они, похоже, больше похожи на дроссели, чтобы предотвратить повторную обработку; то, что я хочу, является дросселем, который предотвращает повторное создание одних и тех же заданий. Таким образом, объект всегда обрабатывается в самом свежем состоянии. Есть ли для этого плагин или техника?

Обновление: у меня не было времени на создание промежуточного программного обеспечения, но у меня была связанная функция очистки, чтобы обеспечить уникальность очередей: https://gist.github.com/mahemoff/bf419c568c525f0af903 p >

Ответ 1

Мое предложение - искать ранее запланированные задания на основе некоторых критериев выбора и удалять, прежде чем планировать новый. Это было полезно для меня, когда я хочу одно запланированное задание для определенного объекта и/или один из его методов.

Некоторые примеры методов в этом контексте:

 find_jobs_for_object_by_method(klass, method)

  jobs = Sidekiq::ScheduledSet.new

  jobs.select { |job|
    job.klass == 'Sidekiq::Extensions::DelayedClass' &&
        ((job_klass, job_method, args) = YAML.load(job.args[0])) &&
        job_klass == klass &&
        job_method == method
  }

end

##
# delete job(s) specific to a particular class,method,particular record
# will only remove djs on an object for that method
#
def self.delete_jobs_for_object_by_method(klass, method, id)

  jobs = Sidekiq::ScheduledSet.new
  jobs.select do |job|
    job.klass == 'Sidekiq::Extensions::DelayedClass' &&
        ((job_klass, job_method, args) = YAML.load(job.args[0])) &&
        job_klass == klass &&
        job_method == method  &&
        args[0] == id
  end.map(&:delete)

end

##
# delete job(s) specific to a particular class and particular record
# will remove any djs on that Object
#
def self.delete_jobs_for_object(klass, id)

  jobs = Sidekiq::ScheduledSet.new
  jobs.select do |job|
    job.klass == 'Sidekiq::Extensions::DelayedClass' &&
        ((job_klass, job_method, args) = YAML.load(job.args[0])) &&
        job_klass == klass &&
        args[0] == id
  end.map(&:delete)

end

Ответ 2

Как насчет простого промежуточного ПО клиента?

module Sidekiq
  class UniqueMiddleware

    def call(worker_class, msg, queue_name, redis_pool)
      if msg["unique"]
        queue = Sidekiq::Queue.new(queue_name)
        queue.each do |job|
          if job.klass == msg['class'] && job.args == msg['args']
            return false
          end
        end
      end

      yield

    end
  end
end

Просто зарегистрируйте его

  Sidekiq.configure_client do |config|
    config.client_middleware do |chain|
      chain.add Sidekiq::UniqueMiddleware
    end
  end

Затем в вашей задаче просто установите unique: true в sidekiq_options при необходимости

Ответ 3

Взгляните на это: https://github.com/mhenrixon/sidekiq-unique-jobs

Это sidekiq с добавленными уникальными заданиями

Ответ 4

Возможно, вы могли бы использовать Queue Classic, который запускает задания в базе данных Postgres (по-настоящему открытому), поэтому он может быть расширен (открыт -source), чтобы проверить уникальность, прежде чем делать это.