Alauda Streaming Service for RabbitMQ
  • Русский

    • Восстановление клиентского подключения и failover

    Восстановление RabbitMQ client — это ответственность приложения. Broker может принимать повторные подключения, но приложению по-прежнему нужны корректный выбор конечной точки, повторы, подтверждения и поведение с идемпотентностью.

    Используйте это руководство, чтобы повысить устойчивость producers и consumers к перезапускам node, rolling upgrades, временным сбоям сети и изменениям конечных точек на уровне площадки.

    Содержание

    Design PrinciplesJava ExamplePython ExampleEndpoint Failover StrategyVerification ChecklistRelated Information

    Design Principles

    Используйте следующие принципы проектирования для production-клиентов:

    • Настраивайте более одного адреса broker, если client library это поддерживает.
    • Разделяйте подключения producer и consumer, чтобы одна заблокированная цепочка не влияла на другую.
    • Используйте heartbeats и разумные timeouts подключения.
    • Включайте автоматическое восстановление, если client library предоставляет такую возможность.
    • Используйте publisher confirms для producers.
    • Используйте manual acknowledgements и идемпотентную обработку для consumers.
    • Рассматривайте failover на уровне площадки как изменение конфигурации приложения, даже если включено автоматическое восстановление на уровне node.

    Java Example

    RabbitMQ Java client поддерживает автоматическое восстановление подключения и topology recovery:

    ConnectionFactory factory = new ConnectionFactory();
factory.setUsername("<username>");
factory.setPassword("<password>");
factory.setVirtualHost("/");
factory.setAutomaticRecoveryEnabled(true);
factory.setTopologyRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(5000);
factory.setRequestedHeartbeat(30);
factory.setConnectionTimeout(10000);

Address[] addresses = {
    new Address("<ip-1>", <port-1>),
    new Address("<ip-2>", <port-2>),
    new Address("<ip-3>", <port-3>)
};

Connection connection = factory.newConnection(addresses);
Channel channel = connection.createChannel();
channel.confirmSelect();

    Используйте publisher confirms на channels producer и проверяйте ошибки подтверждения в коде приложения.

    Python Example

    С pika реализуйте явный цикл повторного подключения и заново создавайте channels или consumers после сбоев подключения:

    import pika
import random
import time

def process_message(body):
    print(f"Processing {body!r}")

def on_message(channel, method, properties, body):
    try:
        process_message(body)
        channel.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

credentials = pika.PlainCredentials("<username>", "<password>")
endpoints = [
    pika.ConnectionParameters("<ip-1>", <port-1>, "/", credentials, heartbeat=30),
    pika.ConnectionParameters("<ip-2>", <port-2>, "/", credentials, heartbeat=30),
    pika.ConnectionParameters("<ip-3>", <port-3>, "/", credentials, heartbeat=30),
]

while True:
    try:
        random.shuffle(endpoints)
        connection = pika.BlockingConnection(endpoints)
        channel = connection.channel()
        channel.basic_qos(prefetch_count=50)
        channel.basic_consume(
            queue="orders",
            on_message_callback=on_message,
            auto_ack=False,
        )
        channel.start_consuming()
    except pika.exceptions.AMQPConnectionError:
        time.sleep(5)
        continue

    Если ваш consumer зависит от объявленной topology, воссоздавайте или проверяйте topology после повторного подключения в соответствии с поведением вашей client library.

    Endpoint Failover Strategy

    Используйте одну из следующих стратегий выбора endpoint:

    СтратегияКогда использоватьПримечания
    Несколько адресов broker в clientПриложения могут получать несколько адресов и переключаться между ними.Хорошо подходит для failover на уровне node внутри одного cluster.
    Адрес Kubernetes Service или LoadBalancerКлиенты должны использовать один стабильный адрес.Упрощает конфигурацию, но убедитесь, что тип service соответствует источнику traffic.
    Переключение конфигурации между primary и DR endpointsВам нужен failover на уровне площадки.Используйте вместе с документированным operational runbook и логикой перезапуска или reload приложения.

    Автоматическое восстановление между node в одном cluster не переключает приложение автоматически на другую DR-площадку. Failover на уровне площадки по-прежнему требует изменений конфигурации, DNS или service discovery.

    Verification Checklist

    После изменения настроек восстановления client проверьте, что:

    • Приложение может подключаться более чем к одному адресу broker.
    • Publishers используют confirms и явно завершаются с ошибкой, если публикации не принимаются.
    • Consumers используют manual acknowledgement и могут корректно перезапускаться.
    • Цикл повторного подключения не создает дубликаты регистраций consumer.
    • Настройки TLS остаются действительными для каждого endpoint, который может использовать client.