# 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 等消息队列系统奠定了基础。