Message Queue User Guide ​

This page walks you through the practical usage of the Message Queue feature in EMQX, from creating queues to configuring their behavior and managing them using the Dashboard, REST API, or configuration files.

Manually Create Message Queue via Dashboard ​

Message Queues must be explicitly declared/created before they can store or dispatch messages. You can create message queues either manually or automatically. For details about automatic creation, see Automatically Create Message Queue via Dashboard.

To create a new Message Queue manually using the EMQX Dashboard:

1. Navigate to Message Queue in the left menu.

2. Click the Create button on the page.

3. In the Create Message Queue dialog, configure the following options:

   • Topic Filter: Enter the topic or topic filter (e.g., t/1). It defines which published messages are enqueued based on topic matching. A queue will collect all messages that match this topic filter.

     To consume messages from the queue, clients must subscribe to the topic using the $q/{Topic Filter} format.

   • Dispatch Strategy: Select how messages should be distributed among subscribers. Available strategies include:

     • Least Inflight Subscriber: Prefer subscribers with the fewest unacknowledged messages.
     • Random: (default) Select a subscriber at random.
     • Round Robin: Rotate delivery evenly across all subscribers.

   • Data Retention Period: Specify how long messages should be retained in the queue. You can set the time unit (e.g., days).

   • Last Value Semantics: This option is enabled by default. When enabled, a new message with the same queue key will overwrite any previous, unconsumed message with that key in the same queue. This ensures only the most recent message per key is retained. The default key is the client ID of the message publisher.

     • Queue Key Expression: When Last-Value Semantics is enabled, this field defines the expression used to extract the key from each message. The default value is message.from, which means the client ID of the message publisher. This field supports configuration using Variform expressions.

   • Max Shard Message Count: (Optional) Sets the maximum number of messages allowed in each shard of the queue. You can toggle this setting on and enter a custom value, or leave it disabled to allow unlimited messages (infinity). This setting is persisted to durable storage.

   • Max Shard Message Bytes: (Optional) Sets the maximum total size (in bytes) of messages in each shard of the queue. You can toggle this setting on and enter a value (e.g., 200MB), or leave it disabled for unlimited size (infinity). This setting is persisted to durable storage.

     Performance Note

     Queues with size limits may have slower write performance, especially under high throughput conditions.

4. Click Create to save the queue.

The new queue will appear in the Message Queue list, showing its topic filter, dispatch strategy, last-value semantics status, and data retention period. You can edit queue settings or delete queues using the buttons in the Actions column.

Queue Key Expression ​

The Queue Key Expression specifies how to extract the key used for message deduplication in Last-Value Semantics mode. This expression is evaluated against a message's metadata and follows the syntax of Variform expressions.

The expression is evaluated against a message context that includes fields such as from, topic, payload, headers.properties, and more. For example, to use a user property as the key, you could set the expression to:

message.headers.properties.'User-Property'.user-prop

If the key cannot be extracted based on the expression (e.g., the field doesn't exist), the message will be discarded and not enqueued.

Message Context Example ​

Queue Key Expressions are evaluated against the following message structure:

{
  "message": {
    "qos": 0,
    "topic": "some/topic",
    "payload": "some-payload",
    "headers": {
      "client_attrs": {},
      "proto_ver": 5,
      "properties": {
        "User-Property": {
          "user-prop": "some-value"
        }
      },
      "peerhost": "127.0.0.1",
      "username": "undefined",
      "protocol": "mqtt",
      "peername": "127.0.0.1:49352"
    },
    "from": "clientid",
    "timestamp": 1759238376252,
    "id": "..non utf8 bytes...",
    "flags": {
      "retain": false,
      "dup": false
    },
    "extra": {}
  }
}

#{message =>
      #{extra => #{},
        flags => #{dup => false, retain => false},
        id => <<0,6,64,4,154,125,229,77,244,69,0,0,28,21,0,2>>,
        timestamp => 1759238376252, from => <<"clientid">>,
        headers =>
            #{peername => <<"127.0.0.1:49352">>, protocol => mqtt,
              username => undefined, peerhost => <<"127.0.0.1">>,
              properties =>
                  #{'User-Property' => #{<<"user-prop">> => <<"some-value">>}},
              proto_ver => 5, client_attrs => #{}
            },
        payload => <<"some-payload">>, topic => <<"some/topic">>,
        qos => 0
      }
    }

Automatically Create Message Queue via Dashboard ​

Starting from EMQX 6.0.1, Message Queues can be automatically created when clients subscribe to a $q/-prefixed topic. This allows queues to be provisioned dynamically without manual setup.

The queues may be auto-created either as regular queues or last-value semantics queues.

Auto Create Last Value Semantics Queue ​

This option is turned on by default in the Message Queue tab under MQTT Settings. It allows EMQX to automatically create queues that support Last-Value Semantics, where only the most recent message with a given key is retained.

1. Navigate to Management -> MQTT Settings -> Message Queue tab.

2. By default, Enable Auto Create Last Value Semantics Queue is enabled.

   Configure the following:

   • Queue Key Expression: Required. Defines how to extract a unique key from each message (default: message.from).
   • Dispatch Strategy: Determines how messages are distributed to subscribers (default: Random).
   • Data Retention Period: Specifies how long messages should be retained in the queue.

3. Click Save Changes.

When a client subscribes to a topic such as $q/test, EMQX will automatically create a last-value semantics queue, which will appear in the Message Queue list.

Auto Create Regular Message Queue ​

This option can be enabled manually if you prefer regular queues where messages are stored independently and not overwritten.

1. Go to Management -> MQTT Settings -> Message Queue tab.
2. Turn on Enable Auto Create Regular Message Queue.
3. Configure the following:
   • Dispatch Strategy: Determines how messages are distributed to subscribers (default: Random).
   • Data Retention Period: Specifies how long messages should be retained in the queue.
4. Click Save Changes.

Configure Message Queue Settings ​

This section explains how to configure global settings that apply to all Message Queues in EMQX. These settings control message retention, cleanup intervals, internal queue behavior, and queue auto-creation behavior. You can configure them via the Dashboard, REST API, or configuration file.

Dashboard ​

You can update Message Queue settings directly from the EMQX Dashboard without restarting the broker. This is useful for making changes to system-wide behavior at runtime.

To configure global settings for Message Queues via the Dashboard:

1. Go to Management -> MQTT Settings -> Message Queue tab.

   Alternatively, you can click the Settings button in the top-right corner of the Message Queue page.

2. In the Message Queue panel, the following configuration options are available:

   • Enable Message Queue: The message queue system is enabled by default and cannot be disabled via the Dashboard.

     To disable it, you must modify the configuration file directly.

   • Max Queue Count: Sets the maximum number of queues that can be created.

   • GC Interval: The interval at which expired messages are cleaned up from queues. Default is 1 hour.

   • Regular Queue Retention Period: The maximum duration for which messages are retained in regular queues. Default is 7 days.

   • Find Queue Retry Interval: When a client subscribes to a $q/-prefixed queue topic and the corresponding queue does not yet exist, this setting controls how often the client retries to find the queue. Default is 10 seconds.

   • Auto-Creation Options: EMQX supports dynamic queue provisioning through auto-creation features.

     • Auto Create Last Value Semantics Queue (enabled by default): When a client subscribes to a $q/ topic and no matching queue exists, EMQX will automatically create a queue with Last-Value Semantics enabled.

       For details of the settings, see Auto Create Last Value Semantics Queue.

     • Auto Create Regular Message Queue: Can be enabled as an alternative to the above. When enabled, EMQX will create regular (non-overwriting) queues automatically for $q/ subscriptions.

       For details of the settings, see Auto Create Regular Message Queue.

3. After making changes, click Save Changes to apply the new settings.

REST API ​

You can also configure global Message Queue settings via the REST API. These settings apply system-wide and affect how all queues are managed internally.

curl -v -u key:secret -X PUT -H "Content-Type: application/json" http://localhost:18083/api/v5/message_queues/config -d '{"find_queue_retry_interval": "10s", "gc_interval": "1h", "regular_queue_retention_period": "7d"}'

Configuration File ​

For persistent and version-controlled configuration, you can define Message Queue settings in the EMQX configuration file (emqx.conf). Below is an example with key settings:

mq {
    gc_interval = 1h
    regular_queue_retention_period = 1d
    find_queue_retry_interval = 10s
    max_queue_count = 100
    }
}

Configuration Descriptions ​

• gc_interval: Defines the interval at which the Message Queues will clean up expired messages.
• regular_queue_retention_period: Sets the maximum time that messages are retained in a regular queue. After this period, messages will be purged.
• find_queue_retry_interval: Determines how frequently a subscriber retries to locate a queue when subscribing to a $q/ topic that does not yet exist.
• max_queue_count: (Optional) Sets the maximum number of queues that can be created.

Manage Message Queue via REST API ​

EMQX provides a set of REST APIs to manage the lifecycle of Message Queues, including creation, retrieval, update, and deletion.

Create a Message Queue ​

Create a new message queue by specifying the topic filter and queue properties such as whether to enable Last-Value Semantics:

curl -s -u key:secret -X POST -H "Content-Type: application/json" \
http://localhost:18083/api/v5/message_queues \
-d '{"topic_filter": "t1/#", "is_lastvalue": false, "limits": {"max_shard_message_count": 10000, "max_shard_message_bytes": "200MB"}}' | jq

List All Message Queues ​

Retrieve the list of existing message queues:

curl -s -u key:secret -X GET -H "Content-Type: application/json" \
http://localhost:18083/api/v5/message_queues | jq

Update a Message Queue ​

Update the properties of an existing queue, such as its dispatch strategy:

curl -s -u key:secret -X PUT -H "Content-Type: application/json" \
http://localhost:18083/api/v5/message_queues/t1%2F%23 \
-d '{"dispatch_strategy": "least_inflight", "limits": {"max_shard_message_count": 5000, "max_shard_message_bytes": "100MB"}}' | jq

Delete a Message Queue ​

Remove a message queue and all messages retained in it:

curl -s -u key:secret -X DELETE \
http://localhost:18083/api/v5/message_queues/t1%2F%23

Note:

• Topic filters in the URL must be URL-encoded (e.g., t1/# becomes t1%2F%23).
• Authentication is required (key:secret).

FAQ and Troubleshooting ​

Why aren't messages being enqueued? ​

• Make sure the topic filter of the declared Message Queue matches the topic of the published message.
• Verify that the queue exists and is properly configured.
• Check the EMQX logs for relevant errors or warnings. Look specifically for entries with the mq_ prefix to diagnose queue-related issues.

What happens when queues exceed capacity? ​

Message Queues in EMQX now support multiple types of capacity limits. If any of these limits are reached, EMQX will remove the oldest messages during garbage collection (GC) until the queue size returns within the configured bounds.

• Time-based limit: All queues are still subject to the configured retention period. Once messages exceed the retention period, they are no longer eligible for delivery and will be automatically purged during GC.

• Size-based limits: You can optionally configure per-shard limits on:

  • Max number of messages (max_shard_message_count)
  • Max total size of messages in bytes (max_shard_message_bytes)

  These limits are soft and applied during GC, not in real time. Queues may temporarily exceed the configured thresholds between GC cycles.

  Note that these limits apply per shard in durable storage. For information on how to configure the number of shards, see Number of Shards. In addition, size limits do not account for the replication factor; the actual physical storage used by a queue will be multiplied by the replication factor.