Очереди заданий на списках

Это руководство строит очередь заданий на списке Redis: от простой очереди до надёжной, которая возвращает задания остановившегося воркера.

Пример - сервис изображений. Из одной очереди он прогревает кеш страниц, а из другой строит миниатюры.

Простая очередь

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "cache.warmups")]
struct Warmup {
    page: String,
}

#[subscriber(RedisList::new("cache.warmups"))]
async fn warm_cache(warmup: &Warmup) -> HandlerOutcome {
    tracing::info!(warmup.page, "page cached");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Warmup {
    page: String,
}

outgoing!(Warmup, "cache.warmups");

struct WarmCache;

impl Handle<Warmup> for WarmCache {
    fn handle(
        &self,
        warmup: &Warmup,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(warmup.page, "page cached");
        ready(Ok(()))
    }
}

Производитель добавляет запись в левый конец списка командой LPUSH. RedisList::new извлекает записи с правого конца командой BRPOP, поэтому записи выходят в том порядке, в каком вошли, и каждая достаётся ровно одному потребителю. Все работающие экземпляры сервиса извлекают записи из одного списка и так делят работу между собой.

Извлечение удаляет запись из Redis до запуска обработчика. Если сервис останавливается во время работы обработчика, запись пропадает. Очередь доставляет каждую запись не более одного раза, и исход обработчика ничего не завершает. Пропущенный прогрев кеша стоит одной медленной страницы, поэтому cache.warmups - простая очередь.

Надёжная очередь

Макросы

src/main.rs

/// Longer than the slowest render ever takes: a claim idle this long belongs to no live worker.
const MIN_IDLE: Duration = Duration::from_secs(60);

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct ThumbnailJob {
    image: String,
}

#[subscriber(
    RedisList::new("{thumbnails}")
        .reliable()
        .min_idle(MIN_IDLE)
        .recovery_zset("{thumbnails}.claims")
)]
async fn render(job: &ThumbnailJob) -> HandlerOutcome {
    tracing::info!(job.image, "thumbnail rendered");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Longer than the slowest render ever takes: a claim idle this long belongs to no live worker.
const MIN_IDLE: Duration = Duration::from_secs(60);

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct ThumbnailJob {
    image: String,
}

outgoing!(ThumbnailJob);

struct Render;

impl Handle<ThumbnailJob> for Render {
    fn handle(
        &self,
        job: &ThumbnailJob,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(job.image, "thumbnail rendered");
        ready(Ok(()))
    }
}

fn thumbnails() -> RedisList {
    RedisList::new("{thumbnails}")
        .reliable()
        .min_idle(MIN_IDLE)
        .recovery_zset("{thumbnails}.claims")
}

reliable не извлекает запись, а одной командой BLMOVE переносит её в список обработки. Список обработки называется {thumbnails}.processing: это ключ очереди с суффиксом .processing, если processing не задаёт другое имя. Запись остаётся там, пока работает обработчик:

• ack удаляет её из списка обработки командой LREM;
• retry переносит её обратно в очередь;
• drop удаляет её.

Поэтому задание покидает Redis только после того, как обработчик с ним закончил, и очередь доставляет каждую запись хотя бы один раз.

Возврат задания остановившегося воркера

Воркер, который остановился между переносом и подтверждением, оставляет свою запись в списке обработки. Список не хранит, кто и когда взял запись, поэтому сама она не возвращается.

Для этого recovery_zset включает сторожевой механизм. Каждый перенос записывает запись в упорядоченное множество с оценкой, равной времени переноса, а подтверждение удаляет её оттуда. Во время чтения подписка возвращает в очередь каждую запись, которая простояла дольше min_idle, и её может взять любой воркер. С множеством для восстановления min_idle обязателен, и подписка без него не запускается. Задайте его больше самого долгого времени выполнения задания: меньшее значение отбирает задание у воркера, который ещё его выполняет, и миниатюра строится дважды. recovery_ttl задаёт срок жизни брошенного множества для восстановления и должен быть больше min_idle.

Оценки - это миллисекунды настенных часов, поэтому часы хостов сервиса должны совпадать.

Наполнение очереди

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "uploads")]
struct Upload {
    image: String,
}

// Braces in `#[outgoing(name = ..)]` read as a name template, so the mount site names the key.
#[subscriber(RedisStream::new("uploads").group("thumbnailer"), reply("{thumbnails}"))]
async fn queue_thumbnail(upload: &Upload) -> ThumbnailJob {
    ThumbnailJob {
        image: upload.image.clone(),
    }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Upload {
    image: String,
}

outgoing!(Upload, "uploads");

struct QueueThumbnail;

impl Handle<Upload, ThumbnailJob> for QueueThumbnail {
    fn handle(
        &self,
        upload: &Upload,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<ThumbnailJob, HandlerOutcome>> {
        ready(Ok(ThumbnailJob {
            image: upload.image.clone(),
        }))
    }
}

queue_thumbnail читает загрузки из потока и отвечает на каждую заданием на миниатюру. Задание уходит в {thumbnails}, имя которого задаёт точка монтирования. Имя в #[outgoing(name = ..)] читает фигурные скобки как шаблон имени, поэтому ключ с хеш-тегом задаётся в точке монтирования.

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("images", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(warm_cache);
            b.include(render);
            // A queue nobody fed for a day expires, with whatever is left in it.
            b.include(queue_thumbnail)
                .out_reply(list::Publish::new().ttl(Duration::from_hours(24)));
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("images", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(subscriber(RedisList::new("cache.warmups"), WarmCache).build());
            b.include(subscriber(thumbnails(), Render).build());
            // A queue nobody fed for a day expires, with whatever is left in it.
            b.include(
                subscriber(
                    RedisStream::new("uploads").group("thumbnailer"),
                    QueueThumbnail,
                )
                .reply()
                .to("{thumbnails}")
                .build(),
            )
            .out_reply(list::Publish::new().ttl(Duration::from_hours(24)));
        },
    )
}

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

Ответ уходит через list::Publish, который добавляет запись командой LPUSH. Его ttl заново ставит срок жизни ключа очереди при каждом добавлении. Очередь, которой пользуются, не истекает, а очередь, которую никто не пополнял сутки, исчезает вместе со всем, что в ней ещё лежит. У списка нет срока жизни отдельной записи, поэтому срок всегда распространяется на всю очередь.

Очереди в кластере

В кластере надёжная очередь и её список обработки должны находиться в одном хеш-слоте, потому что одна команда переносит запись между ними. Общий слот даёт хеш-тег: Redis хеширует только часть ключа в фигурных скобках, поэтому {thumbnails} и {thumbnails}.processing попадают на один узел. Надёжная очередь, два ключа которой попадают в разные слоты, не запускается, а ошибка называет хеш-тег, который нужно использовать. Пример уже задаёт ключи с тегом, поэтому переносится в кластер без изменений.

Когда вместо списка нужен поток

Список не хранит историю: обработанная запись исчезает, и второй сервис уже не может её прочитать. Сторожевой механизм восстановления - единственная повторная доставка у списка, и собственного счётчика доставок у списка нет. Если одну работу читают несколько сервисов, если работе нужно повторное чтение или если задание остановившегося воркера должно вернуться без сторожевого механизма, используйте поток с группой потребителей, как показывает руководство Группы потребителей и восстановление после сбоя.

Тесты в файле примера публикуют прогрев и загрузку и проверяют, что получили обработчики. Один из них сдвигает часы дальше min_idle и проверяет, что выполненное задание не возвращается.