RabbitMQ — Полное руководство по брокеру сообщений

version: '3.8'

services:
rabbitmq:
image: rabbitmq:3-management
container_name: rabbitmq
ports:
  - "5672:5672"    # AMQP
  - "15672:15672"  # Management UI
environment:
  - RABBITMQ_DEFAULT_USER=admin
  - RABBITMQ_DEFAULT_PASS=secret
volumes:
  - rabbitmq_data:/var/lib/rabbitmq

volumes:
rabbitmq_data:

import pika
import json

# Подключение к RabbitMQ
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# Создаём очередь (если не существует)
channel.queue_declare(queue='orders', durable=True)

# Отправляем сообщение
order = {
'order_id': 12345,
'product': 'iPhone 15',
'quantity': 2,
'user_id': 67890
}

channel.basic_publish(
exchange='',  # default exchange
routing_key='orders',
body=json.dumps(order),
properties=pika.BasicProperties(
    delivery_mode=2  # persistent
)
)

print(f"✅ Sent order: {order['order_id']}")
connection.close()

import pika
import json

connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost')
)
channel = connection.channel()

channel.queue_declare(queue='orders', durable=True)

def process_order(ch, method, properties, body):
order = json.loads(body)
print(f"📦 Processing order: {order['order_id']}")
print(f"   Product: {order['product']}")
print(f"   Quantity: {order['quantity']}")

# Имитация обработки
import time
time.sleep(1)

# Подтверждаем обработку
ch.basic_ack(delivery_tag=method.delivery_tag)
print(f"✅ Order {order['order_id']} processed!")

# Получаем по одному сообщению за раз
channel.basic_qos(prefetch_count=1)

# Подписываемся на очередь
channel.basic_consume(
queue='orders',
on_message_callback=process_order
)

print('🐰 Waiting for orders. Press CTRL+C to exit')
channel.start_consuming()

const amqp = require('amqplib');

async function sendOrder() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();

const queue = 'orders';

// Создаём очередь
await channel.assertQueue(queue, { durable: true });

const order = {
    order_id: 12345,
    product: 'MacBook Pro',
    quantity: 1,
    user_id: 67890
};

// Отправляем
channel.sendToQueue(
    queue,
    Buffer.from(JSON.stringify(order)),
    { persistent: true }
);

console.log(`✅ Sent: ${order.order_id}`);

await channel.close();
await connection.close();
}

sendOrder().catch(console.error);

const amqp = require('amqplib');

async function consumeOrders() {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    const queue = 'orders';

    await channel.assertQueue(queue, { durable: true });

    // Получаем по одному сообщению
    channel.prefetch(1);

    console.log('🐰 Waiting for orders...');

    channel.consume(queue, async (msg) => {
        if (msg !== null) {
            const order = JSON.parse(msg.content.toString());

            console.log(`📦 Processing: ${order.order_id}`);
            console.log(`   Product: ${order.product}`);

            // Имитация обработки
            await new Promise(r => setTimeout(r, 1000));

            // Подтверждаем
            channel.ack(msg);
            console.log(`✅ Done: ${order.order_id}`);
        }
    });
}

consumeOrders().catch(console.error);

package main

import (
    "encoding/json"
    "log"

    amqp "github.com/rabbitmq/amqp091-go"
)

type Order struct {
    OrderID  int    `json:"order_id"`
    Product  string `json:"product"`
    Quantity int    `json:"quantity"`
    UserID   int    `json:"user_id"`
}

func main() {
    conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
    defer conn.Close()

    ch, _ := conn.Channel()
    defer ch.Close()

    // Объявляем очередь
    q, _ := ch.QueueDeclare(
        "orders", // name
        true,     // durable
        false,    // auto-delete
        false,    // exclusive
        false,    // no-wait
        nil,      // args
    )

    order := Order{
        OrderID:  12345,
        Product:  "AirPods Pro",
        Quantity: 3,
        UserID:   67890,
    }

    body, _ := json.Marshal(order)

    ch.Publish(
        "",     // exchange
        q.Name, // routing key
        false,  // mandatory
        false,  // immediate
        amqp.Publishing{
            DeliveryMode: amqp.Persistent,
            ContentType:  "application/json",
            Body:         body,
        },
    )

    log.Printf("✅ Sent order: %d", order.OrderID)
}

import pika
import json

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# Создаём Topic Exchange
channel.exchange_declare(
    exchange='events',
    exchange_type='topic',
    durable=True
)

# Отправляем разные события
events = [
    ('order.created.premium', {'order_id': 1, 'type': 'premium'}),
    ('order.created.standard', {'order_id': 2, 'type': 'standard'}),
    ('order.shipped.premium', {'order_id': 1, 'status': 'shipped'}),
    ('payment.completed.premium', {'order_id': 1, 'amount': 999}),
    ('user.registered.standard', {'user_id': 42}),
]

for routing_key, event in events:
    channel.basic_publish(
        exchange='events',
        routing_key=routing_key,
        body=json.dumps(event)
    )
    print(f"📤 Sent [{routing_key}]: {event}")

connection.close()

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

channel.exchange_declare(
    exchange='events',
    exchange_type='topic',
    durable=True
)

# Создаём временную очередь
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

# Паттерны подписки из аргументов
# Примеры:
# python topic_consumer.py "order.*.*"       - все события заказов
# python topic_consumer.py "*.*.premium"     - все premium события
# python topic_consumer.py "#"               - все события

binding_keys = sys.argv[1:] or ['#']

for binding_key in binding_keys:
    channel.queue_bind(
        exchange='events',
        queue=queue_name,
        routing_key=binding_key
    )
    print(f"🔗 Bound to: {binding_key}")

def callback(ch, method, properties, body):
    print(f"📥 [{method.routing_key}] {body.decode()}")

channel.basic_consume(
    queue=queue_name,
    on_message_callback=callback,
    auto_ack=True
)

print('🐰 Waiting for events...')
channel.start_consuming()

┌─────────────────────────────────────────────────────────────────┐
│                    E-Commerce с RabbitMQ                        │
└─────────────────────────────────────────────────────────────────┘

[Web API] ──publish──► orders.created
                              │
                    ┌─────────┴─────────┐
                    ▼                   ▼
           [Payment Service]    [Inventory Service]
                    │                   │
                    ▼                   ▼
           orders.paid         orders.reserved
                    │                   │
                    └─────────┬─────────┘
                              ▼
                    [Shipping Service]
                              │
                              ▼
                    orders.shipped
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
    [Email Service]  [SMS Service]  [Analytics]

Exchanges:
  • orders (topic) - все события заказов
  • notifications (fanout) - рассылка во все каналы

Queues:
  • payment.process (binding: orders.created)
  • inventory.reserve (binding: orders.created)  
  • shipping.prepare (binding: orders.paid AND orders.reserved)
  • email.send (binding: orders.*)
  • analytics.track (binding: #)