What is RabbitMQ

RabbitMQ is an open-source message broker software that implements the Advanced Message Queuing Protocol (AMQP). Developed in Erlang, its core function is to decouple system components, buffer messages, and facilitate asynchronous communication.

It supports multi-language development, including Python, and is easy to deploy in a Windows environment, making it an ideal choice for local development and small applications.

Core Features (Windows Environment Adaptation)

1. Native Windows Support: Provides a dedicated installation package that runs without complex configuration.
2. Python Friendly: Quickly integrates via the pika library, with a simple and easy-to-use API.
3. High Reliability: Supports features like persistence, manual acknowledgment, and dead-letter queues to ensure message safety.
4. Flexible Routing: Three core routing types – Direct/Topic/Fanout, to meet different scenarios.
5. Visual Management: Comes with a web management interface for easy monitoring of queues and exchanges.

Typical Application Scenarios

• Decoupling services in local development environments (e.g., local debugging of order and notification modules).
• Asynchronous communication for small Windows services (e.g., log reporting for desktop applications).
• Traffic simulation and stress testing in development environments.

Core Components and Working Principles of RabbitMQ

Core Components

• Producer: The end that sends messages via Python scripts (e.g., local order generation scripts).
• Consumer: The end that receives and processes messages via Python scripts (e.g., local payment callback handling scripts).
• Queue: A container for storing messages, which by default is stored on the local disk in a Windows environment.
• Exchange: The routing hub for messages, distributing them to queues based on routing keys.
• Binding: The rules that connect exchanges and queues, including routing key configurations.
• Virtual Host: A unit for permission isolation, with the default being / to meet local development needs.

Workflow

1. The producer connects to the local RabbitMQ on Windows via Python code and sends messages to the exchange.
2. The exchange delivers messages to the corresponding queue based on routing rules.
3. The consumer listens to the queue, retrieves messages, processes them, and sends an acknowledgment signal after processing.
4. RabbitMQ deletes the message from the queue after receiving the acknowledgment.

Installation and Configuration of RabbitMQ in Windows Environment

Installing Dependencies (Erlang)

RabbitMQ is based on Erlang, so the Erlang environment must be installed first:

1. Download the Erlang installation package: Visit the Erlang official website and select the Windows 64-bit installation package (recommended version 23.3 or higher).
2. Double-click to install, using the default path (e.g., <span>C:\Program Files\Erlang OTP</span>).
3. Verify installation: Open CMD and enter erl -version; if the version number is displayed, the installation is successful.

Installing RabbitMQ

1. Download the RabbitMQ installation package: Visit the RabbitMQ official website and select the Windows installation package (.exe format).
2. Double-click to install, using the default path (e.g., <span>C:\Program Files\RabbitMQ Server</span>), the installation process will automatically associate with the Erlang environment.
3. Start the RabbitMQ service:

• Open CMD (run as administrator).
• Change to the RabbitMQ script directory:<span>cd C:\Program Files\RabbitMQ Server\rabbitmq_server-3.12.7\sbin</span> (adjust the version number according to the actual installation).
• Start the service: rabbitmq-service start (a prompt “Service started successfully” indicates success).
• Set to start on boot: rabbitmq-service enable.

Basic Configuration (Enable Management Interface + Create User)

1. Enable the web management interface (default port 15672):

• Execute in CMD:<span>rabbitmq-plugins enable rabbitmq_management</span>.
• Restart the service: rabbitmq-service restart.

1. Create an admin user (for local development):

• Execute the command to create a user:<span>rabbitmqctl add_user admin 123456</span> (username admin, password 123456).
• Grant admin privileges:<span>rabbitmqctl set_user_tags admin administrator</span>.
• Grant all permissions:<span>rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"</span>.

1. Access the management interface: Open a browser and enter<span>http://localhost:15672</span>, log in with admin/123456 to manage queues and exchanges visually.

Core Operations of RabbitMQ (Windows Python Practical)

Installing Python Dependency Library

Install the pika library (RabbitMQ Python client) via pip in the Windows environment:

pip install pika==1.3.2  # Stable version, suitable for Windows environment

Producer Code (Sending Messages)

import pika

def rabbitmq_producer():
    # 1. Configure connection parameters (default address for local RabbitMQ on Windows is localhost)
    credentials = pika.PlainCredentials(username='admin', password='123456')
    connection_params = pika.ConnectionParameters(
        host='localhost',  # Default address for Windows, no need to modify
        port=5672,         # Default AMQP communication port
        virtual_host='/',
        credentials=credentials
    )

    # 2. Establish connection and channel (with statement automatically closes the connection, no need for manual handling)
    with pika.BlockingConnection(connection_params) as connection:
        channel = connection.channel()

        # 3. Declare exchange (Direct type, persistent configuration)
        channel.exchange_declare(
            exchange='test_exchange',
            exchange_type='direct',
            durable=True  # Persistent, exchange does not get lost after RabbitMQ restarts
        )

        # 4. Send message (enable message persistence)
        routing_key = 'test.routing.key'
        message = 'Hello, RabbitMQ (Windows Python Producer)!'
        channel.basic_publish(
            exchange='test_exchange',
            routing_key=routing_key,
            body=message.encode('utf-8'),  # Message body encoded as byte stream
            properties=pika.BasicProperties(
                delivery_mode=2  # 2 indicates message persistence, avoiding loss on service restart
            )
        )
        print(f"Producer sent successfully: {message}")

if __name__ == "__main__":
    rabbitmq_producer()

Consumer Code (Receiving and Processing Messages)

import pika

def rabbitmq_consumer():
    # 1. Configure connection parameters (same as producer, local address localhost)
    credentials = pika.PlainCredentials(username='admin', password='123456')
    connection_params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        virtual_host='/',
        credentials=credentials
    )

    # 2. Establish connection and channel
    connection = pika.BlockingConnection(connection_params)
    channel = connection.channel()

    # 3. Declare exchange and queue (same as producer to avoid failure due to non-creation)
    channel.exchange_declare(
        exchange='test_exchange',
        exchange_type='direct',
        durable=True
    )
    channel.queue_declare(
        queue='test_queue',
        durable=True,        # Queue persistence
        exclusive=False,     # Non-exclusive, multiple consumers can access
        auto_delete=False    # Do not automatically delete when there are no consumers
    )

    # 4. Bind exchange and queue
    channel.queue_bind(
        queue='test_queue',
        exchange='test_exchange',
        routing_key='test.routing.key'
    )

    # 5. Define message processing callback function
    def message_callback(ch, method, properties, body):
        message = body.decode('utf-8')
        print(f"Consumer received message: {message}")
        # Manually acknowledge the message (ensure business processing is completed before deletion)
        ch.basic_ack(delivery_tag=method.delivery_tag, multiple=False)

    # 6. Configure flow control (process only 1 unacknowledged message at a time)
    channel.basic_qos(prefetch_count=1)

    # 7. Listen to the queue (disable auto-ack, enable manual Ack)
    channel.basic_consume(
        queue='test_queue',
        on_message_callback=message_callback,
        auto_ack=False
    )

    print("Consumer started, listening to the queue (press Ctrl+C to stop)...")
    channel.start_consuming()

if __name__ == "__main__":
    rabbitmq_consumer()

Running Instructions (Windows Environment)

1. First start the producer: run producer.py, the console output “Producer sent successfully” indicates completion.
2. Then start the consumer: run consumer.py, it will automatically receive and print messages, and manually acknowledge after processing.
3. Verify persistence: restart the RabbitMQ service (rabbitmq-service restart), rerun the consumer, and it can still receive unprocessed messages.

Advanced Features in Windows Environment (Python Practical)

Dead Letter Queue (Handling Exception Messages)

Used to store expired, full, or rejected messages, configured to adapt to local development in Windows environment:

import pika

def setup_dead_letter():
    """Configure dead letter queue (Windows local environment)"""
    credentials = pika.PlainCredentials('admin', '123456')
    connection_params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        virtual_host='/'
    )
    
    with pika.BlockingConnection(connection_params) as connection:
        channel = connection.channel()

        # 1. Declare dead letter exchange and dead letter queue
        channel.exchange_declare(
            exchange='dlx_exchange',
            exchange_type='direct',
            durable=True
        )
        channel.queue_declare(
            queue='dlx_queue',
            durable=True
        )
        channel.queue_bind(
            queue='dlx_queue',
            exchange='dlx_exchange',
            routing_key='dlx.routing.key'
        )

        # 2. Declare normal business queue, associate with dead letter exchange
        business_queue_args = {
            'x-dead-letter-exchange': 'dlx_exchange',
            'x-dead-letter-routing-key': 'dlx.routing.key',
            'x-max-length': 5,  # Maximum length of queue 5, excess messages enter dead letter
            'x-message-ttl': 10000  # Message expires in 10 seconds, enters dead letter after expiration
        }
        channel.queue_declare(
            queue='business_queue',
            durable=True,
            arguments=business_queue_args
        )

        # 3. Bind business exchange and queue
        channel.exchange_declare(
            exchange='business_exchange',
            exchange_type='direct',
            durable=True
        )
        channel.queue_bind(
            queue='business_queue',
            exchange='business_exchange',
            routing_key='business.routing.key'
        )

        print("Dead letter queue configuration completed in Windows environment")

def send_business_messages():
    """Send business messages (trigger dead letter conditions)"""
    credentials = pika.PlainCredentials('admin', '123456')
    connection_params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        virtual_host='/'
    )
    
    with pika.BlockingConnection(connection_params) as connection:
        channel = connection.channel()
        # Send 6 messages (queue full, the 6th enters dead letter)
        for i in range(6):
            message = f"Windows business message {i+1}"
            channel.basic_publish(
                exchange='business_exchange',
                routing_key='business.routing.key',
                body=message.encode('utf-8'),
                properties=pika.BasicProperties(delivery_mode=2)
            )
            print(f"Sent message: {message}")

def consume_dead_letters():
    """Consume dead letter messages (Windows local)"""
    credentials = pika.PlainCredentials('admin', '123456')
    connection_params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        virtual_host='/'
    )
    
    connection = pika.BlockingConnection(connection_params)
    channel = connection.channel()

    def dlx_callback(ch, method, properties, body):
        message = body.decode('utf-8')
        print(f"Received from dead letter queue: {message} (Windows environment)")
        ch.basic_ack(delivery_tag=method.delivery_tag)

    channel.basic_consume(
        queue='dlx_queue',
        on_message_callback=dlx_callback,
        auto_ack=False
    )

    print("Dead letter consumer started (Windows)...")
    channel.start_consuming()

# Execution order: 1. Configure → 2. Send messages → 3. Consume dead letters
if __name__ == "__main__":
    setup_dead_letter()
    send_business_messages()
    # consume_dead_letters()  # Run separately

Delay Queue (Local Delayed Tasks on Windows)

Implemented based on TTL + dead letter queue (e.g., automatically process tasks after 30 seconds):

import pika

def setup_delay_queue():
    """Configure delay queue in Windows environment (30 seconds delay)"""
    credentials = pika.PlainCredentials('admin', '123456')
    connection_params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        virtual_host='/'
    )
    
    with pika.BlockingConnection(connection_params) as connection:
        channel = connection.channel()

        # 1. Declare dead letter exchange (receives expired delayed messages)
        channel.exchange_declare(
            exchange='delay_dlx_exchange',
            exchange_type='direct',
            durable=True
        )

        # 2. Declare business queue (process delayed tasks)
        channel.queue_declare(
            queue='delay_business_queue',
            durable=True
        )
        channel.queue_bind(
            queue='delay_business_queue',
            exchange='delay_dlx_exchange',
            routing_key='delay.dlx.key'
        )

        # 3. Declare TTL queue (30 seconds expiration, no consumers)
        ttl_arguments = {
            'x-dead-letter-exchange': 'delay_dlx_exchange',
            'x-dead-letter-routing-key': 'delay.dlx.key',
            'x-message-ttl': 30000  # 30 seconds delay
        }
        channel.queue_declare(
            queue='delay_ttl_queue',
            durable=True,
            arguments=ttl_arguments
        )

        # 4. Bind delay exchange and TTL queue
        channel.exchange_declare(
            exchange='delay_business_exchange',
            exchange_type='direct',
            durable=True
        )
        channel.queue_bind(
            queue='delay_ttl_queue',
            exchange='delay_business_exchange',
            routing_key='delay.routing.key'
        )

        print("Delay queue configuration completed in Windows environment (30 seconds delay)")

def send_delay_message():
    """Send delayed message (e.g., order unpaid notification)"""
    credentials = pika.PlainCredentials('admin', '123456')
    connection_params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        virtual_host='/'
    )
    
    with pika.BlockingConnection(connection_params) as connection:
        channel = connection.channel()
        order_id = "WIN_ORDER_123"
        message = f"Order {order_id} created successfully, will be automatically canceled if not paid after 30 seconds"
        channel.basic_publish(
            exchange='delay_business_exchange',
            routing_key='delay.routing.key',
            body=message.encode('utf-8'),
            properties=pika.BasicProperties(delivery_mode=2)
        )
        print(f"Sent delayed message: {message}")

def consume_delay_message():
    """Consume delayed tasks (Windows local)"""
    credentials = pika.PlainCredentials('admin', '123456')
    connection_params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        virtual_host='/'
    )
    
    connection = pika.BlockingConnection(connection_params)
    channel = connection.channel()

    def delay_callback(ch, method, properties, body):
        message = body.decode('utf-8')
        print(f"Executing delayed task (Windows): {message}")
        ch.basic_ack(delivery_tag=method.delivery_tag)

    channel.basic_consume(
        queue='delay_business_queue',
        on_message_callback=delay_callback,
        auto_ack=False
    )

    print("Delayed task consumer started (Windows)...")
    channel.start_consuming()

# Execution order: 1. Configure → 2. Send messages → 3. Consume delayed tasks
if __name__ == "__main__":
    setup_delay_queue()
    send_delay_message()
    # consume_delay_message()  # Run separately

That’s all for today’s content, I hope it helps you!

Feel free to like, follow, and share.