Load Testing AMQP and RabbitMQ

---

Overview

AMQP is a popular protocol for messaging systems like RabbitMQ. LoadForge supports AMQP testing using the MqttUser interface (via locust-plugins) or a custom pika client in Locust.

Locust Test Script (locust.py) – Using Pika

# locust.py
import pika
from locust import User, between, task

class RabbitUser(User):
    wait_time = between(1, 2)

    def on_start(self):
        params = pika.ConnectionParameters('localhost', 5672)
        self.connection = pika.BlockingConnection(params)
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue='loadforge_test')

    @task
    def publish(self):
        self.channel.basic_publish(exchange='', routing_key='loadforge_test', body='Hello LoadForge')

    @task
    def consume(self):
        method_frame, header_frame, body = self.channel.basic_get(queue='loadforge_test', auto_ack=True)
        if body:
            _ = body

---

Notes:

• Install dependencies: pip install locust pika if you wish to test locally.