Class: AceMQ::AMQP::Patterns::ConsumerGroup

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/patterns/consumer_group.rb

Overview

A group of consumers reading the same queue.

group = Patterns::ConsumerGroup.new(mq, "orders.new", size: 4) do |message|
place(message.payload)
Ack.accept
end

at_exit { group.close }

Two things it saves. Starting workers by hand means remembering to stop every one, and a partial shutdown leaves messages held by a consumer nobody is waiting for. And a group can be sized from configuration, which is the number most often changed after a service is running.

Concurrency, or a group?

concurrency: on Connection#consume runs several handlers on one consumer and one channel. A group runs several consumers, each with its own channel and its own prefetch. Reach for the group when handlers are slow enough that one channel's prefetch becomes the limit, or when a fair share across processes matters: the broker round-robins between consumers, so four here compete evenly with four in another instance where one consumer with concurrency four would not.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection, queue, size:, **options, &handler) ⇒ ConsumerGroup

Returns a new instance of ConsumerGroup.

Parameters:



51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/acemq/amqp/patterns/consumer_group.rb', line 51

def initialize(connection, queue, size:, **options, &handler)
  unless handler
    raise ArgumentError, "a consumer group needs a block to handle messages"
  end
  unless size.to_i.positive?
    raise ArgumentError,
          "a consumer group needs at least one consumer, not #{size}"
  end

  @queue = queue
  @lock = Mutex.new
  @consumers = start(connection, size.to_i, options, handler)
end

Instance Attribute Details

#consumers ⇒ Object (readonly)

Returns the value of attribute consumers.



45
46
47
 
# File 'lib/acemq/amqp/patterns/consumer_group.rb', line 45

def consumers
  @consumers
end

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



45
46
47
 
# File 'lib/acemq/amqp/patterns/consumer_group.rb', line 45

def queue
  @queue
end

Instance Method Details

#close(timeout: Connection::DRAIN_TIMEOUT) ⇒ nil

Stops every consumer and waits for the handlers already running.

Every one is stopped even when one of them refuses, because leaving the rest running after a failed shutdown is worse than the failure. The first refusal is raised once the others are down.

The timeout is for the group, not for each member. A group is the one place this library makes several consumers at once, so a per-consumer wait is multiplied by the size the caller asked for: four consumers at thirty seconds each is two minutes, which is past every orchestrator's grace period and therefore a drain that ends in SIGKILL with everything in flight unsettled. Connection#close is bounded the same way, by the same code.

Parameters:

  • timeout (Numeric) (defaults to: Connection::DRAIN_TIMEOUT) —

    seconds for the whole group

Returns:

  • (nil)

Raises:

  • (DrainTimeout) —

    when the deadline expired with handlers running



85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/acemq/amqp/patterns/consumer_group.rb', line 85

def close(timeout: Connection::DRAIN_TIMEOUT)
  running = @lock.synchronize do
    taken = @consumers
    @consumers = []
    taken
  end

  failure, stranded = Drain.of(running, timeout)
  raise failure if failure
  raise DrainTimeout.new(stranded, timeout) unless stranded.empty?

  nil
end

#size ⇒ Object

How many consumers are running.



66
 
# File 'lib/acemq/amqp/patterns/consumer_group.rb', line 66

def size = @consumers.size