RabbitMQ на Rust: библиотека Lapin

Привет, Хабр!

Сегодня рассмотрим библиотеку Lapin в Rust. Lapin — это библиотека, реализующая протокол AMQP 0.9.1, она помогает взаимодействовать с RabbitMQ.

1. Многоканальная работа: один TCP‑соединение поддерживает множество каналов.

2. Поддержка подтверждений: безопасная обработка сообщений (ack/nack).

3. Интеграция с async: асинхронный API, который вписывается в экосистему Rust.

4. TLS: поддержка защищённых соединений через native-tls, openssl или rustls.

Основные сущности Lapin

Connection

Соединение с RabbitMQ устанавливается один раз и может использоваться для работы с несколькими каналами. Это основа любого взаимодействия.

Пример:

use lapin::{Connection, ConnectionProperties};  let addr = "amqp://user:password@localhost:5672/%2f"; let connection = Connection::connect(&addr, ConnectionProperties::default())     .await     .expect("Ошибка подключения");

Поддерживаются как TCP, так и TLS‑соединения.

Channel

Каналы — это основа взаимодействия с RabbitMQ. Через них создаются очереди, подписки и отправляются сообщения.

Пример:

let channel = connection.create_channel()     .await     .expect("Ошибка создания канала");

RabbitMQ рекомендует использовать отдельные каналы для отправки и получения сообщений. В одном соединении можно создавать множество каналов.

Queue

Очереди — это хранилища сообщений. Они бывают:

• Durable: сохраняются при перезапуске сервера.

• Exclusive: доступны только для текущего соединения.

• Auto‑delete: удаляются, когда больше не используются.

Пример:

use lapin::options::QueueDeclareOptions; use lapin::types::FieldTable;  let queue = channel     .queue_declare(         "task_queue",         QueueDeclareOptions {             durable: true, // Устойчивая очередь             ..Default::default()         },         FieldTable::default(),     )     .await     .expect("Ошибка объявления очереди");

Используйте FieldTable, чтобы настроить TTL сообщений, ограничение размера очереди и т. п.

Exchange

Exchange — это маршрутизатор, который направляет сообщения в очереди на основе типов и правил маршрутизации:

1. Direct: сообщения отправляются в конкретную очередь по ключу маршрутизации.

2. Fanout: сообщение отправляется во все очереди, привязанные к exchange.

3. Topic: сложная маршрутизация по шаблонам.

4. Headers: маршрутизация на основе заголовков сообщения.

Пример настройки exchange и привязки к очереди:

use lapin::options::{ExchangeDeclareOptions, QueueBindOptions};  channel     .exchange_declare(         "my_exchange",         lapin::ExchangeKind::Direct, // Тип exchange         ExchangeDeclareOptions {             durable: true,             ..Default::default()         },         FieldTable::default(),     )     .await     .expect("Ошибка создания exchange");  channel     .queue_bind(         "task_queue",         "my_exchange",         "routing_key", // Ключ маршрутизации         QueueBindOptions::default(),         FieldTable::default(),     )     .await     .expect("Ошибка привязки очереди");

TLS:

Для продакшен‑окружения требуется защищённое соединение. Настроим соединение через rustls:

use lapin::{Connection, ConnectionProperties}; use lapin::tcp::{OwnedTLSConfig, OwnedTLSStream};  let addr = "amqps://user:password@rabbitmq.example.com:5671/"; let tls_config = OwnedTLSConfig::default();  let connection = Connection::connect(     addr,     ConnectionProperties::default().with_tls(tls_config), ) .await .expect("Ошибка подключения через TLS");

Обработка ошибок

Ошибки неизбежны.

1. Используйте retry для повторного подключения или обработки сообщения.

2. Обрабатывайте nack, если сообщение нельзя обработать.

Пример обработки ошибок:

while let Some(delivery) = consumer.next().await {     match delivery {         Ok(delivery) => {             if let Err(err) = process_message(&delivery).await {                 error!("Ошибка обработки сообщения: {:?}", err);                 delivery                     .nack(Default::default())                     .await                     .expect("Ошибка отправки nack");             } else {                 delivery                     .ack(Default::default())                     .await                     .expect("Ошибка отправки ack");             }         }         Err(err) => {             error!("Ошибка получения сообщения: {:?}", err);         }     } }

Пример применения

Напишем приложение, которое:

• Создаёт exchange и очередь.

• Маршрутизирует сообщения.

• Обрабатывает входящие сообщения с QoS и подтверждениями.

use lapin::{     options::{BasicPublishOptions, QueueBindOptions, QueueDeclareOptions},     types::FieldTable,     BasicProperties, Connection, ConnectionProperties, }; use tokio; use tracing::info;  #[tokio::main] async fn main() {     tracing_subscriber::fmt::init();      let addr = "amqp://user:password@localhost:5672/%2f";     let connection = Connection::connect(&addr, ConnectionProperties::default())         .await         .expect("Ошибка подключения");      let channel = connection.create_channel().await.expect("Ошибка создания канала");      channel         .exchange_declare(             "logs",             lapin::ExchangeKind::Fanout,             Default::default(),             FieldTable::default(),         )         .await         .expect("Ошибка создания exchange");      let queue = channel         .queue_declare(             "",             QueueDeclareOptions {                 exclusive: true,                 ..Default::default()             },             FieldTable::default(),         )         .await         .expect("Ошибка объявления очереди");      channel         .queue_bind(             &queue.name(),             "logs",             "",             QueueBindOptions::default(),             FieldTable::default(),         )         .await         .expect("Ошибка привязки очереди");      info!("Очередь привязана к exchange");      tokio::spawn(async move {         for i in 0..5 {             let message = format!("Log message {}", i);             channel                 .basic_publish(                     "logs",                     "",                     BasicPublishOptions::default(),                     message.as_bytes(),                     BasicProperties::default(),                 )                 .await                 .expect("Ошибка отправки сообщения");         }     });      info!("Сообщения отправлены"); }

Создаём exchange типа Fanout, объявляем временную очередь и связываем её с exchange. После этого отправляем несколько сообщений в exchange, которые автоматически маршрутизируются во все привязанные очереди.

Подробнее с библиотекой можно ознакомиться здесь.

Рекомендую обратить внимание на открытые уроки, которые в феврале проведут в Otus преподаватели-практики:

• 11 февраля: «Разбираем анатомию парсера на Rust».
  Разберём устройство игрушечного парсера на Rust, его ключевые компоненты и архитектурные принципы, обеспечивающие надёжность и производительность кода. Записаться

• 17 февраля: «Инцидент-менеджмент в SRE — как быстро найти, устранить и предотвратить сбои в системе».
  Практическое руководство по эффективному управлению аварийными ситуациями в рамках Site Reliability Engineering (SRE). Записаться