The Thread::Queue class implements multi-producer, multi-consumer queues. It is especially useful in threaded programming when information must be exchanged safely between multiple threads. The Thread::Queue class implements all the required locking semantics.
The class implements FIFO (first in, first out) type of queue. In a FIFO queue, the first tasks added are the first retrieved.
Example:
queue = Thread::Queue.new producer = Thread.new do 5.times do |i| sleep rand(i) # simulate expense queue << i puts "#{i} produced" end end consumer = Thread.new do 5.times do |i| value = queue.pop sleep rand(i/2) # simulate expense puts "consumed #{value}" end end consumer.join
# File tmp/rubies/ruby-master/thread_sync.rb, line 56
def initialize(enumerable = nil)
Primitive.queue_initialize(enumerable)
end
Document-method: Queue::new
Creates a new queue instance, optionally using the contents of an enumerable for its initial state.
Example:
q = Thread::Queue.new #=> #<Thread::Queue:0x00007ff7501110d0> q.empty? #=> true q = Thread::Queue.new([1, 2, 3]) #=> #<Thread::Queue:0x00007ff7500ec500> q.empty? #=> false q.pop #=> 1
# File tmp/rubies/ruby-master/thread_sync.rb, line 160
def clear
Primitive.cstmt! %{
queue_clear(queue_ptr(self));
return self;
}
end
Removes all objects from the queue.
# File tmp/rubies/ruby-master/thread_sync.rb, line 123
def close
Primitive.cstmt! %{
if (!queue_closed_p(self)) {
FL_SET_RAW(self, QUEUE_CLOSED);
wakeup_all(&queue_ptr(self)->waitq);
}
return self;
}
end
Closes the queue. A closed queue cannot be re-opened.
After the call to close completes, the following are true:
-
closed?will return true -
closewill be ignored. -
calling enq/push/<< will raise a
ClosedQueueError. -
when
empty?is false, calling deq/pop/shift will return an object from the queue as usual. -
when
empty?is true, deq(false) will not suspend the thread and will return nil. deq(true) will raise aThreadError.
ClosedQueueError is inherited from StopIteration, so that you can break loop block.
Example:
q = Thread::Queue.new Thread.new{ while e = q.deq # wait for nil to break loop # ... end } q.close
# File tmp/rubies/ruby-master/thread_sync.rb, line 138
def closed?
Primitive.cexpr!('RBOOL(FL_TEST_RAW(self, QUEUE_CLOSED))')
end
Returns true if the queue is closed.
# File tmp/rubies/ruby-master/thread_sync.rb, line 155
def empty?
Primitive.cexpr!('RBOOL(queue_ptr(self)->len == 0)')
end
Returns true if the queue is empty.
# File tmp/rubies/ruby-master/thread_sync.rb, line 184
def freeze
raise TypeError, "cannot freeze #{self}"
end
The queue can’t be frozen, so this method raises an exception:
Thread::Queue.new.freeze # Raises TypeError (cannot freeze #<Thread::Queue:0x...>)
# File tmp/rubies/ruby-master/thread_sync.rb, line 147
def length
Primitive.cexpr!('LONG2NUM(queue_ptr(self)->len)')
end
Returns the length of the queue.
# File tmp/rubies/ruby-master/thread_sync.rb, line 171
def num_waiting
Primitive.cexpr!('INT2NUM(queue_ptr(self)->num_waiting)')
end
Returns the number of threads waiting on the queue.
# File tmp/rubies/ruby-master/thread_sync.rb, line 71
def pop(non_block = false, timeout: nil)
if non_block && timeout
raise ArgumentError, "can't set a timeout if non_block is enabled"
end
Primitive.rb_queue_pop(non_block, timeout)
end
Retrieves data from the queue.
If the queue is empty, the calling thread is suspended until data is pushed onto the queue. If non_block is true, the thread isn’t suspended, and ThreadError is raised.
If timeout seconds have passed and no data is available nil is returned. If timeout is 0 it returns immediately.
# File tmp/rubies/ruby-master/thread_sync.rb, line 88
def push(object)
Primitive.cexpr!('queue_do_push(self, queue_ptr(self), object)')
end
Pushes the given object to the queue.