Промышленная эксплуатация

Это руководство настраивает соединение сервиса на RabbitMQ для промышленной эксплуатации и описывает, что сервис делает, когда пропадает соединение или останавливается он сам.

Соединение

Макросы

src/main.rs

/// The broker as production runs it.
fn broker(uri: String) -> LapinBroker {
    LapinBroker::new(uri)
        // What the management UI lists for this service's connection.
        .connection_name(concat!("orders ", env!("CARGO_PKG_VERSION")))
        // At most 64 unacknowledged deliveries per subscription.
        .prefetch(nonzero!(64))
}

fn service(broker: LapinBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(take_order);
    })
}

#[ruststream::app]
fn app() -> impl App {
    // amqps://orders:secret@rabbit.internal:5671/shop in production.
    let uri = env::var("AMQP_URL").unwrap_or_else(|_| "amqp://localhost:5672".to_owned());
    service(broker(uri))
}

Вручную

src/main.rs

/// The broker as production runs it.
fn broker(uri: String) -> LapinBroker {
    LapinBroker::new(uri)
        // What the management UI lists for this service's connection.
        .connection_name(concat!("orders ", env!("CARGO_PKG_VERSION")))
        // At most 64 unacknowledged deliveries per subscription.
        .prefetch(nonzero!(64))
}

fn service(broker: LapinBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(RabbitQueue::new("orders"), TakeOrder).build());
    })
}

fn app() -> RustStream {
    // amqps://orders:secret@rabbit.internal:5671/shop in production.
    let uri = env::var("AMQP_URL").unwrap_or_else(|_| "amqp://localhost:5672".to_owned());
    service(broker(uri))
}

fn main() -> ExitCode {
    run_main(app)
}

URI соединения задаёт всё соединение: схему, учётные данные, хост, порт и виртуальный хост. amqp://orders:secret@rabbit.internal:5672/shop входит под пользователем orders в виртуальный хост shop. Виртуальный хост по умолчанию / записывается как %2f. Параметры запроса, например heartbeat, перечислены в спецификации URI RabbitMQ. Сервис читает URI из переменной AMQP_URL, поэтому пароль не попадает в код.

connection_name подписывает соединение в веб-интерфейсе управления, и оператор видит, какому сервису и какой его версии оно принадлежит. prefetch ограничивает число неподтверждённых доставок каждой подписки. Так медленный сервис сдерживает сервер (Очереди задач).

LapinBroker::new только запоминает URI. Сервис подключается при запуске. Если URI не разбирается или сервер недоступен, запуск прерывается с ошибкой AmqpError::Connect.

TLS

URI со схемой amqps:// подключается по TLS, обычно на порт 5671. TLS включает одна из трёх фич ruststream-lapin, и каждая включает соответствующую фичу lapin:

Фича	Библиотека TLS
tls-native-tls	библиотека платформы: OpenSSL в Linux, Secure Transport в macOS, SChannel в Windows
tls-rustls	rustls
tls-rustls-ring	rustls с криптопровайдером ring

Сертификат сервера проверяется по доверенным корневым сертификатам выбранной библиотеки. Настройку сервера описывает руководство по TLS RabbitMQ. Строки зависимостей показывает компонент TLS.

Документ AsyncAPI сервиса сообщает хост и версию AMQP, но не учётные данные и не виртуальный хост из URI, потому что документ публикуется и передаётся другим.

Топология в промышленной эксплуатации

Промышленный сервис обычно работает с выключенным declare_topology, на очередях, которые создаёт и меняет эксплуатация. Тогда отсутствующая очередь прерывает запуск, и сообщения не теряются (Топология). Предел попыток кворумной очереди и очередь ожидания отложенной переобработки в такой схеме тоже настраиваются на сервере (Подтверждения, переобработка и недоставленные сообщения).

Соединения и каналы

Весь сервис работает через одно соединение. Каждая подписка открывает собственный канал, поэтому ошибка на одной очереди не останавливает остальные. Публикации без подтверждения делят один канал. Издатель с подтверждениями, издатель с транзакциями канала и запрашивающая сторона открывают каждый собственный канал при первом использовании, потому что подтверждения, транзакции и direct reply-to - это состояние канала.

При ошибке уровня канала RabbitMQ закрывает канал, например после публикации в несуществующий обменник. Соединение остаётся открытым, а издатель открывает новый канал при следующем вызове.

Потеря соединения

Когда соединение с RabbitMQ пропадает из-за перезапуска сервера, переключения на резервный узел или закрытия оператором, сервис подключается заново сам. Каждая подписка снова подключается к серверу и заново открывает потребителя с тем же объявлением и той же предвыборкой. Попытки повторяются с растущей паузой: от 100 мс с удвоением до 10 с. Каждая неудачная попытка пишется в журнал предупреждением, а восстановление - информационной строкой. Издатели и запрашивающая сторона подключаются заново при следующем вызове и возвращают ошибку, пока сервер недоступен. Пока соединения нет, состояние сервиса остаётся Running.

Доставки, которые сервис взял и не подтвердил, возвращаются в очередь при разрыве соединения, и RabbitMQ доставляет их снова с пометкой Redelivered. Поэтому обработчик, который работал с сообщением в момент разрыва, может получить его ещё раз. Если сервер отменяет потребителя на открытом канале, потому что его очередь удалили, подписка завершается.

Остановка

SIGTERM или Ctrl+C останавливает сервис. Рантайм останавливает подписки и закрывает соединение штатно, по протоколу. RabbitMQ возвращает в очередь каждую доставку, которую сервис взял и не подтвердил, и её забирает другой экземпляр. Издатель или запрашивающая сторона, пережившие соединение, например в фоновой задаче, возвращают AmqpError::Closed и ничего не отправляют в закрытое соединение.

Тестирование

Тесты в файле примера запускают сервис с URI amqps:// с учётными данными и с URI, записанным с ошибкой. Внутри процесса URI разбирается так же, как его разбирает соединение, поэтому с URI, по которому сервис не смог бы подключиться, не запускается и тест. TLS и потеря соединения требуют сервера, поэтому их проверяет тест на сервере.