# msgq : 基于 Redis Stream 的健壮消息队列

[English](en.md)

基于 Redis Stream 的健壮消息队列，支持自动认领和重试处理。

## 功能特性

- **基于 Redis Stream**: 利用 `XREADGROUP` 高效消费消息。
- **消费组支持**: 支持通过消费组进行扩展式消息处理。
- **自动创建组**: 如果消费组不存在，自动创建。
- **高可靠性**: 内置 `BLOCK` 和 `CLAIM` 机制处理闲置消息。
- **并发处理**: 使用 `tokio::spawn` 并发处理消息。
- **错误处理**: 自动重试机制（最多 3 次），并提供错误回调。

## 使用演示

```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
  let read_group = ReadGroup::new("smtp", "send", "test", 1, 3, 2);

  // 运行消费循环
  read_group.run(print_mail, on_error).await?;
  OK
}
```

## 设计思路

`ReadGroup::run` 方法执行一个持续循环：

1.  构建带有 `BLOCK` 和 `CLAIM` 选项的 `XREADGROUP` 命令。
2.  从指定的 Redis Stream 获取消息。
3.  如果消费组缺失，调用 `auto_new` 自动创建。
4.  使用提供的回调函数生成异步任务处理每条消息。
5.  监控任务执行：
    -   如果任务失败，记录错误。
    -   如果重试超过 3 次，将消息移交错误处理程序。
    -   通过 `rm_err_id_li` 确认并删除（`XACK` + `XDEL`）失败消息。

## 技术堆栈

-   **Rust**: 核心语言。
-   **Tokio**: 异步运行时。
-   **Fred**: Redis 客户端。
-   **xkv**: Redis 连接管理。

## 目录结构

-   `src/lib.rs`: 库导出。
-   `src/read_group.rs`: 核心消费逻辑。
-   `src/auto_new.rs`: 消费组管理。
-   `src/rm_err_id_li.rs`: 错误清理逻辑。
-   `tests/`: 集成测试和示例。

## API 参考

### `ReadGroup`

配置和运行消费者的主要结构体。

-   `new(stream, group, consumer, block_sec, claim_idle_sec, count)`: 创建新实例。
-   `run(parse, on_error)`: 启动处理循环。

### `StreamItem`

表示来自流的消息。

-   `id`: 消息 ID。
-   `kv`: 键值对（负载）。

## 历史背景

Redis Streams 在 Redis 5.0 中引入，为 Redis 带来了日志数据结构，灵感来源于 Kafka。它支持消费组，允许分布式消息处理，多个消费者可以从同一个流中读取，而不会重复处理同一条消息。`XREADGROUP` 是此模式的主要命令，直接在 Redis 内部促进了健壮且可扩展的消息队列架构。