Class: Sidekiq::Workers
- Inherits:
-
Object
- Object
- Sidekiq::Workers
- Includes:
- Enumerable
- Defined in:
- lib/sidekiq/api.rb
Overview
A worker is a thread that is currently processing a job. Programmatic access to the current active worker set.
WARNING WARNING WARNING
This is live data that can change every millisecond. If you call #size => 5 and then expect #each to be called 5 times, you’re going to have a bad time.
workers = Sidekiq::Workers.new
workers.size => 2
workers.each do |process_id, thread_id, work|
# process_id is a unique identifier per Sidekiq process
# thread_id is a unique identifier per thread
# work is a Hash which looks like:
# { 'queue' => name, 'run_at' => timestamp, 'payload' => msg }
# run_at is an epoch Integer.
end
Instance Method Summary collapse
- #each ⇒ Object
-
#size ⇒ Object
Note that #size is only as accurate as Sidekiq’s heartbeat, which happens every 5 seconds.
Instance Method Details
#each ⇒ Object
806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 |
# File 'lib/sidekiq/api.rb', line 806 def each Sidekiq.redis do |conn| procs = conn.smembers('processes') procs.sort.each do |key| valid, workers = conn.pipelined do conn.exists(key) conn.hgetall("#{key}:workers") end next unless valid workers.each_pair do |tid, json| yield key, tid, Sidekiq.load_json(json) end end end end |
#size ⇒ Object
Note that #size is only as accurate as Sidekiq’s heartbeat, which happens every 5 seconds. It is NOT real-time.
Not very efficient if you have lots of Sidekiq processes but the alternative is a global counter which can easily get out of sync with crashy processes.
828 829 830 831 832 833 834 835 836 837 838 839 840 841 |
# File 'lib/sidekiq/api.rb', line 828 def size Sidekiq.redis do |conn| procs = conn.smembers('processes') if procs.empty? 0 else conn.pipelined do procs.each do |key| conn.hget(key, 'busy') end end.map(&:to_i).inject(:+) end end end |