İlk Consumer uygulaması

Bu sayfada orders.created kuyruğundaki JSON mesajlarını okuyup sipariş kimliğini yazdıran bir uygulama kuracağız. Sonraki publisher sayfası aynı sözleşmeyle order-42 mesajını gönderir.

Projeyi oluşturun#

Rust 1.96.1 ve edition 2024 kullanın. Bağımsız bir Cargo projesi oluşturup Cargo.toml dosyasını aşağıdakiyle düzenleyin. Registry'de yayımlanmış lilyrs 0.1.0 yeterlidir; framework repository'sini dependency olarak klonlamanız gerekmez.

cargo new orders-consumer
cd orders-consumer

[package]
name = "orders-consumer"
version = "0.1.0"
edition = "2024"
rust-version = "1.96.1"

[dependencies]
lilyrs = { version = "0.1.0", features = ["consumer"] }
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

consumer, queue tiplerini ve DI desteğini de açar. serde mesajın JSON dönüşümü içindir. tokio bağımlılığı #[tokio::main] runtime'ını kurar; Consumer kendisi yeni bir Tokio runtime oluşturmaz.

Worker ve handler yazın#

Aşağıdaki tam dosyayı indirin ve src/main.rs olarak kaydedin:

use lilyrs::{
    consumer::{Consumer, ConsumerError, Injectable, ServiceTrait},
    queue::{Json, QueueHandlerError, queue, queue_service},
};
use serde::Deserialize;

#[derive(Deserialize)]
struct OrderCreated { order_id: String }

#[derive(Default, Injectable)]
#[service(lifetime = "Singleton")]
struct OrderWorker;
impl ServiceTrait for OrderWorker {}

#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "json")]
    async fn created(&self, Json(event): Json<OrderCreated>) -> Result<(), QueueHandlerError> {
        println!("{}", event.order_id);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<(), ConsumerError> {
    Consumer::run().await
}

OrderWorker DI'ya Singleton olarak kaydolur. #[queue_service] içindeki #[queue], fiziksel kuyruk adını ve kabul edilen version/content çiftini bildirir. Json<OrderCreated> gövdeyi çözer; Ok(()) normal pipeline başarısını bildirir. Worker'lar link sırasında keşfedilir; elle add_handler listesi tutulmaz.

RabbitMQ bağlantısını ve kuyruğu tanımlayın#

Yerel RabbitMQ'nuz 127.0.0.1:5672 adresinde çalışmalı ve guest kullanıcısının / vhost'una erişimine izin vermelidir. Bu örnek yerel development bağlantısıdır; broker kurmaz. Aşağıdaki configuration dosyasını indirin ve projenin köküne lily.toml adıyla koyun:

[rabbitmq.consumer]
hostname = "127.0.0.1"
port = 5672
username = "guest"
password = "guest"
vhost = "/"
use_tls = false
pool_size = 1

[[rabbitmq.topology.queues]]
name = "orders.created"
exchange_name = "orders"
routing_key = "orders.created"
concurrency = 2
prefetch_count = 10
retry_attempts = 3
retry_backoff_millis = 1000
max_retry_backoff_millis = 60000

[rabbitmq.topology.queues.retention]
main_max_messages = 100000
main_max_bytes = 1073741824
retry_bucket_max_messages = 10000
retry_bucket_max_bytes = 268435456
dead_letter_max_messages = 10000
dead_letter_max_bytes = 268435456

Connection bölümü broker'a nasıl ulaşılacağını, topology bölümü orders.created kuyruğunun hangi exchange ve routing key ile kullanılacağını belirtir. Altı retention değeri gereklidir; ana kuyruk, retry bucket'ları ve dead-letter kuyruğuna mesaj ve byte sınırları koyar. Bunlar uygulamanızın kapasitesine göre değiştirilecek örnek değerlerdir.

concurrency = 2 aynı anda yürütülebilen delivery sayısını sınırlar; prefetch_count = 10 broker'ın onay bekleyen delivery penceresidir. İki ayar aynı şey değildir. Ayrıntılar kapasite ve topology sayfalarındadır.

Çalıştırın ve mesaj gönderin#

LILY_CONFIG_PATH=./lily.toml LILY_CONFIG_MODE=development cargo run

Önce Consumer'ın broker'a bağlanmasını ve kuyruk registration'ını tamamlamasını bekleyin. Broker yönetim ekranında orders.created için aktif consumer görebilirsiniz; uygulamadan readiness sorgulamak için managed çalışma kullanılır. Sonra başka terminalde publisher örneğini çalıştırın; bu terminalde order-42 yazdırılır.

cargo check Rust sözleşmesini kontrol eder; broker erişimi ve uçtan uca teslimat için iki programın çalıştırılması gerekir. Tracing varsayılan kapalıdır ve Consumer HTTP portu açmaz. Bu nedenle başlangıç doğrulaması olarak otomatik /health endpoint'i beklemeyin.

Uygulamayı kapatın#

Ctrl+C yeni admission'ı durdurup kabul edilmiş işleri kapanış bütçesi içinde uzlaştırmayı başlatır. Consumer::run().await sonucunu bekleyin. Varsayılan application shutdown bütçesi 30 saniyedir; sonucu değerlendirme ve incomplete cleanup ayrımı kapanış sayfasındadır.