Advanced Consumer Configuration

Athomic provides granular controls to fine-tune how consumers process messages, allowing you to improve performance and ensure that critical data is handled with the necessary urgency.

Priority Queues

In many systems, not all messages carry the same weight. A "password reset" notification, for example, is far more critical than a weekly newsletter dispatch. Priority Queues allow you to ensure that high-priority messages are processed before low-priority ones, regardless of their arrival order.

How It Works

Under the hood, Athomic implements prioritization by publishing messages to distinct physical topics based on their priority. If you publish to a logical topic named notifications, a message with HIGH priority will go to a physical topic like notifications.high, while a NORMAL priority message will go to notifications. This allows you to scale resources (e.g., more pods/workers) specifically for higher-priority topics.

Usage

You can set a consumer's priority using the priority parameter in the @subscribe_to or @ordered_consumer decorators.

from nala.athomic.integration.messaging import subscribe_to
from nala.athomic.types.messaging import MessagePriority

# This handler will consume from a high-priority topic (e.g., 'user-events.high')
@subscribe_to(
    topic="user-events",
    group_id="password-reset-workers",
    priority=MessagePriority.HIGH
)
async def handle_password_reset(event: PasswordResetEvent, **kwargs):
    # Immediate and critical processing
    ...

# This handler will consume from a low-priority topic (e.g., 'user-events.low')
@subscribe_to(
    topic="user-events",
    group_id="newsletter-workers",
    priority=MessagePriority.LOW
)
async def handle_newsletter_signup(event: NewsletterEvent, **kwargs):
    # Background, less critical processing
    ...

To publish a message with a specific priority, use the priority parameter in the publish method:

await producer.publish(
    topic="user-events",
    message=password_reset_data,
    priority=MessagePriority.HIGH
)

Concurrency Control

For resource-intensive tasks (e.g., heavy CPU, database connections), you might want to limit how many messages a single instance of your service processes concurrently. This helps prevent the service from becoming overloaded and ensures stable performance.

The concurrency parameter sets a "ceiling" on parallel executions for a specific handler within a single process.

Usage

Pass the concurrency parameter to the decorator.

# This consumer instance will never execute more than 5 `handle_report` tasks at the same time.
@subscribe_to(
    topic="report-generation-tasks",
    group_id="report-workers",
    concurrency=5
)
async def handle_report_generation(task: ReportTask, **kwargs):
    await do_heavy_io_work()

!!! note "Concurrency vs. Scalability" Concurrency control (concurrency) is a form of vertical scaling within a single instance (pod/process). It limits the resources used by that instance. Horizontal scaling is achieved by increasing the number of instances (replicas/pods) of your service. Both techniques can and should be used together for optimal performance.

Runtime Tuning (handler_tuning)

One of Athomic's most powerful features is the ability to override decorator settings (priority and concurrency) at runtime through your settings.toml file. This allows SRE and DevOps teams to fine-tune application performance in production without requiring code changes or new deployments.

How It Works

You can add [[integration.messaging.handler_tuning]] sections to your settings file. Each entry is an "array of tables" that identifies a handler by its unique handler_id and defines the new values for its parameters.

The handler_id is the function's fully qualified import path (e.g., my_app.consumers.my_handler).

Complete Example

Imagine the following code in my_app/consumers.py:

# my_app/consumers.py
from nala.athomic.integration.messaging import subscribe_to
from nala.athomic.types.messaging import MessagePriority

@subscribe_to(
    topic="user-events",
    group_id="password-reset-workers",
    priority=MessagePriority.NORMAL, # Default: NORMAL
    concurrency=10                  # Default: 10
)
async def handle_password_reset(event: PasswordResetEvent, **kwargs):
    ...

Now, in a production environment, we want to treat password resets as high priority and increase the concurrency. We can do this in settings.toml:

# settings.toml

[[integration.messaging.handler_tuning]]
handler_id = "my_app.consumers.handle_password_reset"
priority = "high"
concurrency = 25

With this configuration, even though the code defines the priority as NORMAL and concurrency as 10, Athomic will start this consumer at runtime with priority=HIGH and concurrency=25.