[English](#en) | [中文](#zh)
---
<a id="en"></a>
# msgq : Robust Redis Stream Message Queue
Robust Redis Stream based message queue with auto-claim and retry handling.
## Table of Contents
- [Features](#features)
- [Usage](#usage)
- [Design](#design)
- [Tech Stack](#tech-stack)
- [Directory Structure](#directory-structure)
- [API Reference](#api-reference)
- [History](#history)
## Features
- **Redis Stream Based**: Utilizes `XREADGROUP` for efficient message consumption.
- **Consumer Groups**: Supports scalable message processing with consumer groups.
- **Auto Group Creation**: Automatically creates consumer groups if they don't exist.
- **Reliability**: Built-in `BLOCK` and `CLAIM` mechanisms for handling idle messages.
- **Concurrency**: Processes messages concurrently using `tokio::spawn`.
- **Error Handling**: Automatic retry mechanism (configurable limit) with error callbacks.
## Usage
```rust
use std::sync::Arc;
use aok::{OK, Void};
use log::info;
use msgq::{Kv, ReadGroup};
async fn on_error(mail: Arc<Kv>, error: String) -> Void {
info!("on_error {error} {:?}", mail);
OK
}
async fn print_mail(mail: Arc<Kv>) -> Void {
info!("print_mail {:?}", mail);
OK
}
#[tokio::main]
async fn main() -> Void {
// Initialize Redis connection (example)
// xboot::init().await?;
// Create a ReadGroup
// stream: "smtp", group: "send", consumer: "test"
// block: 1s, claim_idle: 3s, count: 2, max_retry: 3
let read_group = ReadGroup::new("smtp", "send", "test", 1, 3, 2, 3);
// Run the consumer loop
read_group.run(print_mail, on_error).await?;
OK
}
```
## Design
The `ReadGroup::run` method executes a continuous loop that ensures robust message processing:
1. **Command Construction**: Prepares an `XREADGROUP` command with `BLOCK` (for waiting on new messages) and `CLAIM` (for recovering idle messages from other consumers) options.
2. **Message Fetching**: Executes the command against Redis to retrieve a batch of messages.
3. **Group Management**: Checks if the consumer group exists; if not, `auto_new` is called to create it automatically.
4. **Concurrent Processing**: Iterates through the fetched messages and spawns a `tokio` task for each message to execute the user-provided `parse` callback.
5. **Error Handling & Retry**:
- Monitors the execution of each task.
- If a task fails, the error is logged.
- The system tracks retries; if a message fails more than `max_retry` times, it is handed off to the `on_error` callback for manual intervention or logging.
6. **Cleanup**: Successfully processed messages (or those handled by `on_error`) are acknowledged and deleted from the stream using `rm_id_li` (`XACK` + `XDEL`) to prevent reprocessing.
## Tech Stack
- **Rust**: Core language for performance and safety.
- **Tokio**: Asynchronous runtime for handling concurrency.
- **Fred**: High-performance Redis client.
- **xkv**: Redis connection and command management wrapper.
## Directory Structure
- `src/lib.rs`: Library exports and type definitions.
- `src/read_group.rs`: Core `ReadGroup` struct and consumer loop logic.
- `src/auto_new.rs`: Logic for automatically creating consumer groups if missing.
- `src/rm_id_li.rs`: Helper for acknowledging and deleting processed messages.
- `src/parse_stream.rs`: Utilities for parsing Redis Stream responses.
- `tests/`: Integration tests demonstrating usage patterns.
## API Reference
### `ReadGroup`
The main struct for configuring and running the consumer.
- `new(stream, group, consumer, block_sec, claim_idle_sec, count, max_retry)`: Creates a new `ReadGroup` instance.
- `stream`: Redis Key for the stream.
- `group`: Consumer group name.
- `consumer`: Consumer name.
- `block_sec`: Seconds to block waiting for new messages.
- `claim_idle_sec`: Seconds before claiming idle messages from other consumers.
- `count`: Maximum number of messages to fetch per batch.
- `max_retry`: Maximum number of retries before passing to `on_error`.
- `run(parse, on_error)`: Starts the infinite processing loop.
- `parse`: Async callback function to process each message.
- `on_error`: Async callback function to handle messages that failed retries.
### `StreamItem`
Represents a single message from the Redis Stream.
- `id`: The unique ID of the message.
- `retry`: The delivery count (number of times delivered).
- `idle_ms`: Time in milliseconds the message has been idle (if claimed).
- `kv`: The message payload as a vector of Key-Value byte pairs.
## History
In 1983, Vivek Ranadive, a 26-year-old MIT graduate, observed that while hardware components communicated via a "bus", software lacked a similar standard mechanism. He envisioned a "software bus" where applications could publish and subscribe to information without direct, rigid connections. This idea led to the creation of **The Information Bus (TIB)**, the first commercial message queue software. TIB revolutionised financial trading floors by replacing manual chalkboards with real-time digital streams, allowing different trading systems to communicate instantly. This innovation laid the groundwork for modern event-driven architectures and the message queue systems we rely on today, such as Redis Streams.
---
## About
This project is an open-source component of [js0.site ⋅ Refactoring the Internet Plan](https://js0.site).
We are redefining the development paradigm of the Internet in a componentized way. Welcome to follow us:
* [Google Group](https://groups.google.com/g/js0-site)
* [js0site.bsky.social](https://bsky.app/profile/js0site.bsky.social)
---
<a id="zh"></a>
# msgq : 基于 Redis Stream 的健壮消息队列
基于 Redis Stream 的健壮消息队列，支持自动认领和重试处理。
## 目录
- [功能特性](#功能特性)
- [使用演示](#使用演示)
- [设计思路](#设计思路)
- [技术堆栈](#技术堆栈)
- [目录结构](#目录结构)
- [API 参考](#api-参考)
- [历史背景](#历史背景)
## 功能特性
- **基于 Redis Stream**: 利用 `XREADGROUP` 高效消费消息。
- **消费组支持**: 支持通过消费组进行扩展式消息处理。
- **自动创建组**: 若消费组不存在，自动创建。
- **高可靠性**: 内置 `BLOCK` 和 `CLAIM` 机制处理闲置消息。
- **并发处理**: 使用 `tokio::spawn` 并发处理消息。
- **错误处理**: 自动重试机制（可配置重试次数），并提供错误回调。
## 使用演示
```rust
use std::sync::Arc;
use aok::{OK, Void};
use log::info;
use msgq::{Kv, ReadGroup};
async fn on_error(mail: Arc<Kv>, error: String) -> Void {
info!("on_error {error} {:?}", mail);
OK
}
async fn print_mail(mail: Arc<Kv>) -> Void {
info!("print_mail {:?}", mail);
OK
}
#[tokio::main]
async fn main() -> Void {
// 初始化 Redis 连接 (示例)
// xboot::init().await?;
// 创建 ReadGroup
// stream: "smtp", group: "send", consumer: "test"
// block: 1s, claim_idle: 3s, count: 2, max_retry: 3
let read_group = ReadGroup::new("smtp", "send", "test", 1, 3, 2, 3);
// 运行消费循环
read_group.run(print_mail, on_error).await?;
OK
}
```
## 设计思路
`ReadGroup::run` 方法执行一个持续循环，确保消息处理的健壮性：
1. **命令构建**: 构建带有 `BLOCK`（等待新消息）和 `CLAIM`（认领其他消费者的闲置消息）选项的 `XREADGROUP` 命令。
2. **消息获取**: 向 Redis 执行命令以获取一批消息。
3. **组管理**: 检查消费组是否存在；若不存在，调用 `auto_new` 自动创建。
4. **并发处理**: 遍历获取的消息，为每条消息生成一个 `tokio` 异步任务，执行用户提供的 `parse` 回调。
5. **错误处理与重试**:
- 监控每个任务的执行。
- 若任务失败，记录错误日志。
- 系统跟踪重试次数；若消息失败超过 `max_retry` 次，将其移交 `on_error` 回调进行人工干预或记录。
6. **清理**: 成功处理（或由 `on_error` 处理）的消息将通过 `rm_id_li`（`XACK` + `XDEL`）进行确认和删除，防止重复处理。
## 技术堆栈
- **Rust**: 核心语言，提供高性能和安全性。
- **Tokio**: 异步运行时，处理并发任务。
- **Fred**: 高性能 Redis 客户端。
- **xkv**: Redis 连接与命令管理封装。
## 目录结构
- `src/lib.rs`: 库导出与类型定义。
- `src/read_group.rs`: 核心 `ReadGroup` 结构体与消费循环逻辑。
- `src/auto_new.rs`: 消费组自动创建逻辑。
- `src/rm_id_li.rs`: 消息确认与删除辅助工具。
- `src/parse_stream.rs`: Redis Stream 响应解析工具。
- `tests/`: 集成测试与使用示例。
## API 参考
### `ReadGroup`
配置和运行消费者的主要结构体。
- `new(stream, group, consumer, block_sec, claim_idle_sec, count, max_retry)`: 创建新的 `ReadGroup` 实例。
- `stream`: Redis Stream 键名。
- `group`: 消费组名称。
- `consumer`: 消费者名称。
- `block_sec`: 等待新消息的阻塞秒数。
- `claim_idle_sec`: 认领闲置消息前的等待秒数。
- `count`: 每批次获取的最大消息数。
- `max_retry`: 移交 `on_error` 前的最大重试次数。
- `run(parse, on_error)`: 启动无限处理循环。
- `parse`: 处理每条消息的异步回调函数。
- `on_error`: 处理重试失败消息的异步回调函数。
### `StreamItem`
表示来自 Redis Stream 的单条消息。
- `id`: 消息唯一 ID。
- `retry`: 投递次数（重试计数）。
- `idle_ms`: 消息闲置时间（毫秒，若被认领）。
- `kv`: 消息负载，为键值对字节向量。
## 历史背景
1983 年，26 岁的麻省理工学院毕业生 Vivek Ranadive 观察到，虽然硬件组件通过“总线”进行通信，但软件缺乏类似的标准机制。他设想了一种“软件总线”，应用程序可以在没有直接、僵化连接的情况下发布和订阅信息。这一想法促成了 **The Information Bus (TIB)** 的诞生，这是首款商业消息队列软件。TIB 取代了人工黑板，实现了实时数字流，彻底改变了金融交易大厅，允许不同的交易系统即时通信。这一创新为现代事件驱动架构以及我们今天所依赖的 Redis Stream 等消息队列系统奠定了基础。
---
## 关于
本项目为 [js0.site ⋅ 重构互联网计划](https://js0.site) 的开源组件。
我们正在以组件化的方式重新定义互联网的开发范式，欢迎关注：
* [谷歌邮件列表](https://groups.google.com/g/js0-site)
* [js0site.bsky.social](https://bsky.app/profile/js0site.bsky.social)