Tell Us What You Think!¶

Please take a moment to tell us what you think about this guide: Send an e-mail or raise an issue on Github.

Let us know what was unclear or what has not been covered. Maybe you do not like the guide style or grammar or discover spelling mistakes. Reader feedback is key to making the documentation better.

Demo-Consumer and Producer¶

Open up 2 terminals.

Then run the demo consumer

$ php public/index.php humus amqp consumer demo-consumer

Next, open another shell and run the demo producer

$ php public/index.php humus amqp stdin-producer demo-producer "demo-message"

You should see the output in the demo consumer’s shell. It should look something like this:

hallo
2014-08-27T18:43:30+02:00 DEBUG (7): Acknowledged 1 messages at 0 msg/s

If you run the command multiple times, you can see that it the consumer will also ack bundles of messages. You noticed perhabs, that you run it with the stdin-producer command, but what does this mean? Try this:

$ cat README.md | xargs -0 php public/index.php humus amqp stdin-producer demo-producer
$ echo "my test message" | xargs -0 php public/index.php humus amqp stdin-producer demo-producer

For now, let’s check what a demo consumer looks like and how to configure it.

The EchoCallback is the implementation part of the consumer. As you can see, you simply provide a callable, you get the parameters (message and queue) and you’re ready to start. You don’t need to extend or even write yourself the consumer implementation.

The required connection configuration can be found at: module.config.php#L85-L95.

The required exchange configuration is also there: module.config.php#L27-L37.

The required queue configuration: module.config.php#L56-L64.

The required consumer configuration: module.config.php#L112-L119.

And finally, the required producer configuration: module.config.php#L98-L105.

More information about the configuration of the Humus AMQP module, check the other sections of the manual.

That’s it, send a SIGUSR1-signal (kill -10) to stop the consumer. You probably noticed, that there is an error-exchange configured for the demo exchange.

That’s a nice exercise: Go and change the consumer callback to “return false;”, so the messages get a nack and will be routed to the error exchange. Attach a queue to that exchange and create the consumer configuration. You can also reuse the already existing EchoErrorCallback.

Topic consumer and producer example¶

First, run the consumer again:

$ php public/index.php humus amqp consumer topic-consumer-error

This consumer is only interested in routing keys matching #.err, so let’s send some messages with different routing keys.

$ php public/index.php humus amqp stdin-producer topic-producer --route=level.err err
$ php public/index.php humus amqp stdin-producer topic-producer --route=level.warn warn
$ php public/index.php humus amqp stdin-producer topic-producer --route=level.info info
$ php public/index.php humus amqp stdin-producer topic-producer --route=level.debug debug

As you can see, only the first message in interessting for the consumer, all others are trashed. Go, send a lot of messages:

$ php public/index.php humus amqpdemo topic-producer 1000

This will send 1000 messages that will be consumed by the topic-consumer-error. You probably noticed, that by default, the consumer will never ack more than 3 messages at once, even if you send tons of messages. You can change that, go to the module.config.php file:

'topic-consumer-error' => array(
     'queues' => array(
         'info-queue',
     ),
     'callback' => 'HumusAmqpDemoModule\Demo\EchoCallback',
     'qos' => array(
         'prefetch_count' => 100
     ),
     'auto_setup_fabric' => true
 ),

If you set the prefetch count to 100, the consumer will ack up to 100 messages at once. For more information, see: Consumer Prefetch.

Running RPC-client & -server example¶

Open up 3 terminals.

Then run 2 rpc-servers

$ php public/index.php humus amqp rpc-server demo-rpc-server
$ php public/index.php humus amqp rpc-server demo-rpc-server2

Before we start the client, let’s see, how a rpc-server get’s configured what in the demo servers.

First, we need exchanges: module.config.php#L46-L53.

Queues, too: module.config.php#L65-L76.

And here’s how the servers/ clients are configured: module.config.phpL128-L145.

You can check the callbacks here.

The PowerOfTwoCallback does a sleep(1) before returning, the RandomIntCallback does a sleep(2); With this, it’s more easy to show real parallel processing.

Start the rpc-client

$ php public/index.php humus amqpdemo rpc-client 5

This will send 5 messages, you can see in the server output, that the messages are acknowledged and the response in the client afterwards. No let’s send messages to both:

$ php public/index.php humus amqpdemo rpc-client 5 --parallel

What? Don’t believe it? It’s truly parallel!

$ time php public/index.php humus amqpdemo rpc-client 5 --parallel

Enjoy!

See Running from CLI get know more about Humus AMQP Module’s CLI commands.

What are AMQP exchanges?¶

An exchange accepts messages from a producer application and routes them to message queues. They can be thought of as the “mailboxes” of the AMQP world. Unlike some other messaging middleware products and protocols, in AMQP, messages are not published directly to queues. Messages are published to exchanges that route them to queue(s) using pre-arranged criteria called bindings.

There are multiple exchange types in the AMQP 0.9.1 specification, each with its own routing semantics. Custom exchange types can be created to deal with sophisticated routing scenarios (e.g. routing based on geolocation data or edge cases) or just for convenience.

Concept of Bindings¶

A binding is an association between a queue and an exchange. A queue must be bound to at least one exchange in order to receive messages from publishers. Learn more about bindings in the Bindings Guide.

Exchange attributes¶

Exchanges have several attributes associated with them:

• Name
• Type (direct, fanout, topic, headers or some custom type)
• Durability
• Whether the exchange is auto-deleted when no longer used
• Other metadata (sometimes known as X-arguments)

How fanout exchanges route messages¶

A fanout exchange routes messages to all of the queues that are bound to it and the routing key is ignored. If N queues are bound to a fanout exchange, when a new message is published to that exchange a copy of the message is delivered to all N queues. Fanout exchanges are ideal for the broadcast routing of messages.

Graphically this can be represented as:

fanout exchange routing

Declaring a fanout exchange¶

You can simply define an exchange in the module configuration.

<?php

return array(
    'humus_amqp_module' => array(
        'exchanges' => array(
            'myexchange' => array(
                'name' => 'myexchange',
                'type' => 'fanout'
            )
        )
    )
);

Fanout use cases¶

Because a fanout exchange delivers a copy of a message to every queue bound to it, its use cases are quite similar:

• Massively multiplayer online (MMO) games can use it for leaderboard updates or other global events
• Sport news sites can use fanout exchanges for distributing score updates to mobile clients in near real-time
• Distributed systems can broadcast various state and configuration updates
• Group chats can distribute messages between participants using a fanout exchange (although AMQP does not have a built-in concept of presence, so XMPP may be a better choice)

Pre-declared fanout exchanges¶

AMQP 0.9.1 brokers must implement a fanout exchange type and pre-declare one instance with the name of "amq.fanout".

Applications can rely on that exchange always being available to them. Each vhost has a separate instance of that exchange, it is not shared across vhosts for obvious reasons.

How direct exchanges route messages¶

A direct exchange delivers messages to queues based on a message routing key, an attribute that every AMQP v0.9.1 message contains.

Here is how it works:

• A queue binds to the exchange with a routing key K
• When a new message with routing key R arrives at the direct exchange, the exchange routes it to the queue if K = R

A direct exchange is ideal for the unicast routing of messages (although they can be used for multicast routing as well).

Here is a graphical representation:

direct exchange routing

Declaring a direct exchange¶

<?php

return array(
    'humus_amqp_module' => array(
        'exchanges' => array(
            'myexchange' => array(
                'name' => 'myexchange',
                'type' => 'direct'
            )
        )
    )
);

Direct routing example¶

Since direct exchanges use the message routing key for routing, message producers need to specify it:

The routing key will then be compared for equality with routing keys on bindings, and consumers that subscribed with the same routing key each get a copy of the message.

Direct Exchanges and Load Balancing of Messages¶

Direct exchanges are often used to distribute tasks between multiple workers (instances of the same application) in a round robin manner. When doing so, it is important to understand that, in AMQP 0.9.1, messages are load balanced between consumers and not between queues.

The Queues and Consumers guide provides more information on this subject.

Pre-declared direct exchanges¶

AMQP 0.9.1 brokers must implement a direct exchange type and pre-declare two instances:

• amq.direct
• “” exchange known as default exchange (unnamed, referred to as an empty string)

Applications can rely on those exchanges always being available to them. Each vhost has separate instances of those exchanges, they are not shared across vhosts for obvious reasons.

Default exchange¶

The default exchange is a direct exchange with no name pre-declared by the broker. It has one special property that makes it very useful for simple applications, namely that every queue is automatically bound to it with a routing key which is the same as the queue name.

For example, when you declare a queue with the name of “search.indexing.online”, RabbitMQ will bind it to the default exchange using “search.indexing.online” as the routing key. Therefore a message published to the default exchange with routing key = “search.indexing.online” will be routed to the queue “search.indexing.online”. In other words, the default exchange makes it seem like it is possible to deliver messages directly to queues, even though that is not technically what is happening.

Direct Exchange Use Cases¶

Direct exchanges can be used in a wide variety of cases:

• Direct (near real-time) messages to individual players in an MMO game
• Delivering notifications to specific geographic locations (for example, points of sale)
• Distributing tasks between multiple instances of the same application all having the same function, for example, image processors
• Passing data between workflow steps, each having an identifier (also consider using headers exchange)
• Delivering notifications to individual software services in the network

How Topic Exchanges Route Messages¶

Topic exchanges route messages to one or many queues based on matching between a message routing key and the pattern that was used to bind a queue to an exchange. The topic exchange type is often used to implement various publish/subscribe pattern variations.

Topic exchanges are commonly used for the multicast routing of messages.

Topic exchanges can be used for broadcast routing, but fanout exchanges are usually more efficient for this use case.

Declaring a topic exchange¶

<?php

return array(
    'humus_amqp_module' => array(
        'exchanges' => array(
            'myexchange' => array(
                'name' => 'myexchange',
                'type' => 'topic',
                'connection' => 'my_other_connection'
            )
        )
    )
);

As you can see, you can also specify to which connection the exchange belongs to. If nothing is present, the default connection will be used. You can set the default like this:

<?php

return array(
    'humus_amqp_module' => array(
        'default_connection' => 'my_other_connection'
    )
);

Topic Exchange Routing Example¶

Two classic examples of topic-based routing are stock price updates and location-specific data (for instance, weather broadcasts). Consumers indicate which topics they are interested in (think of it like subscribing to a feed for an individual tag of your favourite blog as opposed to the full feed).

A routing pattern consists of several words separated by dots, in a similar way to URI path segments being joined by slash. A few of examples:

• asia.southeast.thailand.bangkok
• sports.basketball
• usa.nasdaq.aapl
• tasks.search.indexing.accounts

The following routing keys match the “americas.south.#” pattern:

• americas.south
• americas.south.brazil
• americas.south.brazil.saopaolo
• americas.south.chile.santiago

In other words, the “#” part of the pattern matches 0 or more words.

For the pattern “americas.south.*”, some matching routing keys are:

• americas.south.brazil
• americas.south.chile
• americas.south.peru

but not

• americas.south
• americas.south.chile.santiago

As you can see, the “*” part of the pattern matches 1 word only.

Topic Exchange Use Cases¶

Topic exchanges have a very broad set of use cases. Whenever a problem involves multiple consumers/applications that selectively choose which type of messages they want to receive, the use of topic exchanges should be considered. To name a few examples:

• Distributing data relevant to specific geographic location, for example, points of sale
• Background task processing done by multiple workers, each capable of handling specific set of tasks
• Stocks price updates (and updates on other kinds of financial data)
• News updates that involve categorization or tagging (for example, only for a particular sport or team)
• Orchestration of services of different kinds in the cloud
• Distributed architecture/OS-specific software builds or packaging where each builder can handle only one architecture or OS

Data serialization¶

You are encouraged to take care of data serialization before publishing (i.e. by using JSON, Thrift, Protocol Buffers or some other serialization library). Note that because AMQP is a binary protocol, text formats like JSON largely lose their advantage of being easy to inspect as data travels across the network, so if bandwidth efficiency is important, consider using MessagePack or Protocol Buffers.

A few popular options for data serialization are:

• JSON
• BSON
• Message Pack
• XML
• Protocol Buffers

Message attributes¶

RabbitMQ messages have various metadata attributes that can be set when a message is published. Some of the attributes are well-known and mentioned in the AMQP 0.9.1 specification, others are specific to a particular application. Well-known attributes are listed here as options that Humus AMQP Module takes:

• :persistent
• :delivery_mode
• :mandatory
• :timestamp
• :expiration
• :type
• :reply_to
• :content_type
• :content_encoding
• :correlation_id
• :priority
• :cluster_id
• :user_id
• :app_id
• :message_id

All other attributes can be added to a headers table.

An example:

<?php

$now = microtime(1);

$attribs = new MessageAttributes()
$attribs->setAppId('amqp.example');
$attribs->setAppId(8);
$attribs->setType('kinda.checkin';
$attribs->setHeaders(array(
    'latitude' => 59.35,
    'longituide' => 18.0666667
));
$attribs->setTimestamp($now)
$attribs->setCorrelationId('r-1');
$attribs->setContentType('application/json');

$producer->publish('{"foo": "bar"}', '', $attribs);

:routing_key

Used for routing messages depending on the exchange type and configuration.

:persistent

When set to true, RabbitMQ will persist message to disk.

:mandatory

This flag tells the server how to react if the message cannot be routed to a queue. If this flag is set to true, the server will return an unroutable message to the producer with a basic.return AMQP method. If this flag is set to false, the server silently drops the message.

:content_type

MIME content type of message payload. Has the same purpose/semantics as HTTP Content-Type header.

:content_encoding

MIME content encoding of message payload. Has the same purpose/semantics as HTTP Content-Encoding header.

:priority

Message priority, from 0 to 9.

:message_id

Message identifier as a string. If applications need to identify messages, it is recommended that they use this attribute instead of putting it into the message payload.

:reply_to

Commonly used to name a reply queue (or any other identifier that helps a consumer application to direct its response). Applications are encouraged to use this attribute instead of putting this information into the message payload.

:correlation_id

ID of the message that this message is a reply to. Applications are encouraged to use this attribute instead of putting this information into the message payload.

:type

Message type as a string. Recommended to be used by applications instead of including this information into the message payload.

:user_id

Sender’s identifier. Note that RabbitMQ will check that the value of this attribute is the same as username AMQP connection was authenticated with, it SHOULD NOT be used to transfer, for example, other application user ids or be used as a basis for some kind of Single Sign-On solution.

:app_id

Application identifier string, for example, “eventoverse” or “webcrawler”

:timestamp

Timestamp of the moment when message was sent, in seconds since the Epoch

:expiration

Message expiration specification as a string

:arguments

A map of any additional attributes that the application needs. Nested hashes are supported. Keys must be strings.

It is recommended that application authors use well-known message attributes when applicable instead of relying on custom headers or placing information in the message body. For example, if your application messages have priority, publishing timestamp, type and content type, you should use the respective AMQP message attributes instead of reinventing the wheel.

Validated User ID¶

In some scenarios it is useful for consumers to be able to know the identity of the user who published a message. RabbitMQ implements a feature known as validated User ID. If this property is set by a publisher, its value must be the same as the name of the user used to open the connection. If the user-id property is not set, the publisher’s identity is not validated and remains private.

Publishing Callbacks and Reliable Delivery in Distributed Environments¶

A commonly asked question about RabbitMQ clients is “how to execute a piece of code after a message is received”.

Message publishing with Bunny happens in several steps:

• AMQPExchange::publish takes a payload and various metadata attributes
• Resulting payload is staged for writing
• On the next event loop tick, data is transferred to the OS kernel using one of the underlying NIO APIs
• OS kernel buffers data before sending it
• Network driver may also employ buffering

As you can see, “when data is sent” is a complicated issue and while methods to flush buffers exist, flushing buffers does not guarantee that the data was received by the broker because it might have crashed while data was travelling down the wire.

The only way to reliably know whether data was received by the broker or a peer application is to use message acknowledgements. This is how TCP works and this approach is proven to work at the enormous scale of the modern Internet. AMQP 0.9.1 fully embraces this fact and Bunny follows.

In cases when you cannot afford to lose a single message, AMQP 0.9.1 applications can use one (or a combination of) the following protocol features:

• Publisher confirms (a RabbitMQ-specific extension to AMQP 0.9.1)
• Publishing messages as mandatory
• Transactions (these introduce noticeable overhead and have a relatively narrow set of use cases)

A more detailed overview of the pros and cons of each option can be found in a blog post that introduces Publisher Confirms extension by the RabbitMQ team. The next sections of this guide will describe how the features above can be used with Bunny.

Publishing messages as mandatory¶

When publishing messages, it is possible to use the :mandatory option to publish a message as “mandatory”. When a mandatory message cannot be routed to any queue (for example, there are no bindings or none of the bindings match), the message is returned to the producer.

Returned messages¶

When a message is returned, the application that produced it can handle that message in different ways:

• Store it for later redelivery in a persistent store
• Publish it to a different destination
• Log the event and discard the message

A returned message handler has access to AMQP method (basic.return) information, message metadata and payload (as a byte array). The metadata and message body are returned without modifications so that the application can store the message for later redelivery.

Publishing Persistent Messages¶

Messages potentially spend some time in the queues to which they were routed before they are consumed. During this period of time, the broker may crash or experience a restart. To survive it, messages must be persisted to disk. This has a negative effect on performance, especially with network attached storage like NAS devices and Amazon EBS. AMQP 0.9.1 lets applications trade off performance for durability, or vice versa, on a message-by-message basis.

To publish a persistent message, use the :persistent option:

<?php

$attribs = new MessageAttributes();
$attribs->setPersistent(true);
$producer->publish($data, '', $attribs);

Note that in order to survive a broker crash, the messages MUST be persistent and the queue that they were routed to MUST be durable.

Durability and Message Persistence provides more information on the subject.

Publishing In Multi-threaded Environments¶

When using Humus AMQP Module in multi-threaded environments, the rule of thumb is: avoid sharing channels across threads.

In other words, publishers in your application that publish from separate threads should use their own channels. The same is a good idea for consumers.

How headers exchanges route messages¶

<?php

return array(
    'humus_amqp_module' => array(
        'exchanges' => array(
            'header-exchange' => array(
                'name' => 'header-exchange',
                'type' => 'headers'
            )
        ),
        'queues' => array(
            'myqueue-1' => array(
                'name' => 'myqueue',
                'exchange' => 'header-exchange',
                'arguments' => array(
                    'os' => 'linux',
                    'x-match' => 'all'
                )
            ),
            'myqueue-2' => array(
                'name' => 'myqueue',
                'exchange' => 'header-exchange',
                'arguments' => array(
                    'os' => 'osx',
                    'x-match' => 'any'
                )
            )
        ),
        'producers' => array(
            'my-producer' => array(
                'exchange' => 'exchanges',
                'auto_setup_fabric' => true
            )
        )
    )
);

<?php

$attribs = new MessageAttributes();

$attribs->setHeaders(array(
    'os' => 'linux',
    'cores' => 8
));
$producer->publish('8 cores/Linux', '', $attribs);

$attribs->setHeaders(array(
    'os' => 'osx',
    'cores' => 8
));

$producer->publish('4 cores/OS X', '', $attribs);

myqueue-2 received 8 cores/Linux

Headers Exchange Routing¶

When there is just one queue bound to a headers exchange, messages are routed to it if any or all of the message headers match those specified upon binding. Whether it is “any header” or “all of them” depends on the "x-match" header value. In the case of multiple queues, a headers exchange will deliver a copy of a message to each queue, just like direct exchanges do. Distribution rules between consumers on a particular queue are the same as for a direct exchange.

Headers Exchange Use Cases¶

Headers exchanges can be looked upon as “direct exchanges on steroids” and because they route based on header values, they can be used as direct exchanges where the routing key does not have to be a string; it could be an integer or a hash (dictionary) for example.

Some specific use cases:

• Transfer of work between stages in a multi-step workflow (routing slip pattern)
• Distributed build/continuous integration systems can distribute builds based on multiple parameters (OS, CPU architecture, availability of a particular package).

Pre-declared Headers Exchanges¶

RabbitMQ implements a headers exchange type and pre-declares one instance with the name of "amq.match". RabbitMQ also pre-declares one instance with the name of "amq.headers". Applications can rely on those exchanges always being available to them. Each vhost has a separate instance of those exchanges and they are not shared across vhosts for obvious reasons.

consistent-hash¶

The consistent hashing AMQP exchange type is a custom exchange type developed as a RabbitMQ plugin. It uses consistent hashing to route messages to queues. This helps distribute messages between queues more or less evenly.

A quote from the project README:

In various scenarios, you may wish to ensure that messages sent to an exchange are consistently and equally distributed across a number of different queues based on the routing key of the message. You could arrange for this to occur yourself by using a direct or topic exchange, binding queues to that exchange and then publishing messages to that exchange that match the various binding keys.

However, arranging things this way can be problematic:

It is difficult to ensure that all queues bound to the exchange will receive a (roughly) equal number of messages without baking in to the publishers quite a lot of knowledge about the number of queues and their bindings.

If the number of queues changes, it is not easy to ensure that the new topology still distributes messages between the different queues evenly.

Consistent Hashing is a hashing technique whereby each bucket appears at multiple points throughout the hash space, and the bucket selected is the nearest higher (or lower, it doesn’t matter, provided it’s consistent) bucket to the computed hash (and the hash space wraps around). The effect of this is that when a new bucket is added or an existing bucket removed, only a very few hashes change which bucket they are routed to.

In the case of Consistent Hashing as an exchange type, the hash is calculated from the hash of the routing key of each message received. Thus messages that have the same routing key will have the same hash computed, and thus will be routed to the same queue, assuming no bindings have changed.

x-random¶

The x-random AMQP exchange type is a custom exchange type developed as a RabbitMQ plugin by Jon Brisbin. A quote from the project README:

It is basically a direct exchange, with the exception that, instead of each consumer bound to that exchange with the same routing key getting a copy of the message, the exchange type randomly selects a queue to route to.

This plugin is licensed under Mozilla Public License 1.1, same as RabbitMQ.

Message Acknowledgements and Their Relationship to Transactions and Publisher Confirms¶

Consumer applications (applications that receive and process messages) may occasionally fail to process individual messages, or might just crash. Additionally, network issues might be experienced. This raises a question - “when should the RabbitMQ remove messages from queues?” This topic is covered in depth in the Queues guide, including prefetching and examples.

In this guide, we will only mention how message acknowledgements are related to AMQP transactions and the Publisher Confirms extension. Let us consider a publisher application (P) that communications with a consumer (C) using AMQP 0.9.1. Their communication can be graphically represented like this:

-----       -----       -----
|   |   S1  |   |   S2  |   |
| P | ====> | B | ====> | C |
|   |       |   |       |   |
-----       -----       -----

We have two network segments, S1 and S2. Each of them may fail. A publisher (P) is concerned with making sure that messages cross S1, while the broker (B) and consumer (C) are concerned with ensuring that messages cross S2 and are only removed from the queue when they are processed successfully.

Message acknowledgements cover reliable delivery over S2 as well as successful processing. For S1, P has to use transactions (a heavyweight solution) or the more lightweight Publisher Confirms, a RabbitMQ-specific extension.

Explicitly Deleting an Exchange¶

Exchanges are deleted using the AMQPExchange#delete:

<?php

/* @var $exchange \AMQPExchange */
$exchange->delete();

Auto-deleted exchanges¶

Exchanges can be auto-deleted. To declare an exchange as auto-deleted, use the :auto_delete option on declaration:

<?php

return array(
    'humus_amqp_module' => array(
        'exchanges' => array(
            'header-exchange' => array(
                'name' => 'header-exchange',
                'type' => 'headers',
                'auto_delete' => true
            )
        ),

An auto-deleted exchange is removed when the last queue bound to it is unbound.

What are AMQP Queues?¶

Queues store and forward messages to consumers. They are similar to mailboxes in SMTP. Messages flow from producing applications to Exchanges that route them to queues and finally, queues deliver the messages to consumer applications (or consumer applications fetch messages as needed).

AMQP 0.9.1 is a programmable protocol, so queues and bindings alike are declared by applications.

Concept of Bindings¶

A binding is an association between a queue and an exchange. Queues must be bound to at least one exchange in order to receive messages from publishers. Learn more about bindings in the Bindings guide.

Queue Attributes¶

Queues have several attributes associated with them:

• Name
• Exclusivity
• Durability
• Whether the queue is auto-deleted when no longer used
• Other metadata (sometimes called X-arguments)

These attributes define how queues can be used, their life-cycle, and other aspects of queue behavior.

Explicitly Named Queues¶

Applications may pick queue names or ask the broker to generate a name for them. The configure a queue with an explicit name:

<?php

return array(
    'humus_amqp_module' => array(
        'exchanges' => array(
            'demo-exchange' => array(
                'name' => 'demo-exchange',
                'type' => 'direct'
            )
        ),
        'queues' => array(
            'my-queue' => array(
                'name' => 'my-queue',
                'exchange' => 'demo-exchange'
            )
        )
    )
);

Server-named queues¶

To ask an AMQP broker to generate a unique queue name for you, pass an empty string as the queue name argument. A generated queue name (like amq.gen-JZ46KgZEOZWg-pAScMhhig) will be assigned to the queue instance that the method returns:

<?php

return array(
    'humus_amqp_module' => array(
        'exchanges' => array(
            'demo-exchange' => array(
                'name' => 'demo-exchange',
                'type' => 'direct'
            )
        ),
        'queues' => array(
            'my-queue' => array(
                'name' => '',
                'exchange' => 'demo-exchange'
            )
        )
    )
);

Reserved Queue Name Prefix¶

Queue names starting with “amq.” are reserved for server-named queues and queues for internal use by the broker. Attempts to declare a queue with a name that violates this rule will result in an AMQPExchangeException with reply code 403 and an exception message similar to this:

Server channel error: 403, message: ACCESS_REFUSED - exchange name 'amq.queue' contains reserved prefix 'amq.*'

This error results in the channel that was used for the declaration being forcibly closed by RabbitMQ. If the program subsequently tries to communicate with RabbitMQ using the same channel without re-opening it then the AMQP Extension will throw an AMQPChannelException' with message 'Could not create exchange. No channel available.

Queue Re-Declaration With Different Attributes¶

When queue declaration attributes are different from those that the queue already has, a channel-level exception with code 406 will be raised. The reply text will be similar to this:

Server channel error: 406, message: PRECONDITION_FAILED - cannot redeclare exchange 'foo' in vhost '/'
with different type, durable, internal or autodelete value

This error results in the channel that was used for the declaration being forcibly closed by RabbitMQ. If the program subsequently tries to communicate with RabbitMQ using the same channel without re-opening it then Bunny will throw an AMQPChannelException' with message 'Could not create exchange. No channel available. In order to continue communications in the same program after such an error, a different channel would have to be used.

How RabbitMQ Routes Messages¶

After a message reaches RabbitMQ and before it reaches a consumer, several things happen:

• RabbitMQ needs to find one or more queues that the message needs to be routed to, depending on type of exchange
• RabbitMQ puts a copy of the message into each of those queues or decides to return the message to the publisher
• RabbitMQ pushes message to consumers on those queues or waits for applications to fetch them on demand

A more in-depth description is this:

• RabbitMQ needs to consult bindings list for the exchange the message was published to in order to find one or more queues that the message needs to be routed to (step 1)
• If there are no suitable queues found during step 1 and the message was published as mandatory, it is returned to the publisher (step 1b)
• If there are suitable queues, a copy of the message is placed into each one (step 2)
• If the message was published as mandatory, but there are no active consumers for it, it is returned to the publisher (step 2b)
• If there are active consumers on those queues and the basic.qos setting permits, message is pushed to those consumers (step 3)

The important thing to take away from this is that messages may or may not be routed and it is important for applications to handle unroutable messages.

Handling of Unroutable Messages¶

Unroutable messages are either dropped or returned to producers. RabbitMQ extensions can provide additional ways of handling unroutable messages: for example, RabbitMQ’s Alternate Exchanges extension makes it possible to route unroutable messages to another exchange. Bunny support for it is documented in the RabbitMQ Extensions guide.

Exchanges and Publishing documentation guide provides more information on the subject, including full code examples.

QoS — Prefetching messages¶

For cases when multiple consumers share a queue, it is useful to be able to specify how many messages each consumer can be sent at once before sending the next acknowledgement. This can be used as a simple load balancing technique to improve throughput if messages tend to be published in batches. For example, if a producing application sends messages every minute because of the nature of the work it is doing.

Imagine a website that takes data from social media sources like Twitter or Facebook during the Champions League (european soccer) final (or the Superbowl), and then calculates how many tweets mentioned a particular team during the last minute. The site could be structured as 3 applications:

• A crawler that uses streaming APIs to fetch tweets/statuses, normalizes them and sends them in JSON for processing by other applications (“app A”).
• A calculator that detects what team is mentioned in a message, updates statistics and pushes an update to the Web UI once a minute (“app B”).
• A Web UI that fans visit to see the stats (“app C”).

In this imaginary example, the “tweets per second” rate will vary, but to improve the throughput of the system and to decrease the maximum number of messages that the AMQP broker has to hold in memory at once, applications can be designed in such a way that application “app B”, the “calculator”, receives 5000 messages and then acknowledges them all at once. The broker will not send message 5001 unless it receives an acknowledgement.

In AMQP 0.9.1 parlance this is known as QoS or message prefetching. Prefetching is configured on a per-channel basis.

Exchange Durability¶

AMQP separates the concept of entity durability (queues, exchanges) from message persistence. Exchanges can be durable or transient. Durable exchanges survive broker restart, transient exchanges do not (they have to be redeclared when the broker comes back online), however, not all scenarios and use cases mandate exchanges to be durable.

To create a durable exchange, declare it with the :durable => true argument.

Queue Durability¶

Queues can be durable or transient. Durable queues survive broker restart, transient queues do not (they have to be redeclared when the broker comes back online), however, not all scenarios and use cases mandate queues to be durable.

To create a durable queue, declare it with the :durable => true argument.

Durability of a queue does not make messages that are routed to that queue durable. If a broker is taken down and then brought back up, durable queues will be re-declared during broker startup, however, only persistent messages will be recovered.

Binding Durability¶

Bindings of durable queues to durable exchanges are automatically durable and are restored after a broker restart. The AMQP 0.9.1 specification states that the binding of durable queues to transient exchanges must be allowed. In this case, since the exchange would not survive a broker restart, neither would any bindings to such and exchange.

Message Persistence¶

Messages may be published as persistent and this, in conjunction with queue durability, is what makes an AMQP broker persist them to disk. If the server is restarted, the system ensures that received persistent messages in durable queues are not lost. Simply publishing a message to a durable exchange or the fact that a queue to which a message is routed is durable does not make that message persistent. Message persistence depends on the persistence mode of the message itself.

Pass the :persistent => true argument to the AMQPExchange#publish method to publish your message as persistent.

Clustering and High Availability¶

To achieve the degree of durability that critical applications need, it is necessary but not enough to use durable queues, exchanges and persistent messages. You need to use a cluster of brokers because otherwise, a single hardware problem may bring a broker down completely.

RabbitMQ offers a number of high availability features for both scenarios with more (LAN) and less (WAN) reliable network connections.

See the RabbitMQ clustering and high availability guides for in-depth discussion of this topic.

Highly Available (Mirrored) Queues¶

Whilst the use of clustering provides for greater durability of critical systems, in order to achieve the highest level of resilience for queues and messages, high availability configuration should be used. This is because although exchanges and bindings survive the loss of individual nodes by using clustering, messages do not. Without mirroring, queue contents reside on exactly one node, thus the loss of a node will cause message loss.

See the RabbitMQ high availability guide for more information about mirrored queues.

Learn More¶

See also rabbitmq.com section on Per-queue Message TTL

How To Use It With Humus AMQP Module¶

The Humus AMQP Module makes already use of the nack method to reject a block of messages. You don’t need to take care of that, unless you want to write a consumer yourself.

Learn More¶

See also rabbitmq.com section on basic.nack

Learn More¶

See also rabbitmq.com section on Alternate Exchanges

Learn More¶

See also rabbitmq.com section on Exchange-to-Exchange Bindings

Learn More¶

See also rabbitmq.com section on Queue Leases

Learn More¶

See also rabbitmq.com section on Per-message TTL

Learn More¶

See also rabbitmq.com section on Sender-Selected Distribution

How To Use It With Bunny 0.9+¶

Dead-letter Exchange is a feature that is used by specifying additional queue arguments:

• "x-dead-letter-exchange" specifies the exchange that dead lettered messages should be published to by RabbitMQ
• "x-dead-letter-routing-key" specifies the routing key that should be used (has to be a constant value)

<?php

return array(
    'humus_amqp_module' => array(
        'queues' => array(
            'foo' => array(
                'name' => 'foo',
                'exchange' => 'demo',
                'arguments' => array(
                    'x-dead-letter-exchange' => 'demo.error'
                ),
        ),
    )
);

Learn More¶

See also rabbitmq.com section on Dead Letter Exchange

Common channel-level exceptions and what they mean¶

A few channel-level exceptions are common and deserve more attention.

Message-Versioning with Routing Keys¶

You can use the routing keys for versioning, like this:

<?php

$producer->publish('some message', 'v1.0.0');

Then you just bind the queue to the routing key. When you update your system, newer messages (e.g. 2.0.0) are not delivered to queue that is not able to process them. On the other hand you can write consumers, that are backwards-compatible, so the queues used will bind to a variety of routing keys (v1.0.0 & v2.0.0 or v1.0.*).

Message-Versioning with extra Attributes¶

Use some custom message headers to specify version constraints:

<?php

$attribs = new MessageAttributes();
$attribs->setHeaders(array(
    'x-version' => 'v1.0.0'
));
$producer->publish('some message', '', $attribs);

This way a message with a version not handled by your consumer, will also be at least delivered to him. The consumer will then need to check for the x-version header and process if possible or throw an exception. Compared to the versioning strategy with routing keys, you’ll always know, that somebody is sending still messages in old version, as you see the exceptions in the log file. If you just deploy a consumer that is only able to acknowlegde messages of e.g. v2.0.0, than it’s hard to recognize that there are still messages in version 1.0.0 going through the system. On the other hand, there are messages send over network, that nobody cares, so routing keys are often preferable.