# redis-queue-executor

**Repository Path**: decentant/redis-queue-executor

## Basic Information

- **Project Name**: redis-queue-executor
- **Description**: 基于 Redis 的轻量级分布式任务队列，支持三种消费模式（轮询 / 广播唤醒 / Stream 消费组），内置任务保活、废弃任务自动重入队、队列背压、节点监控与优雅停机能力。适用于中小规模、无需引入专业消息中间件的分布式任务调度场景。
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-09-27
- **Last Updated**: 2026-09-27

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 分布式队列执行器（queue-execute）

基于 Redis 的轻量级分布式任务队列，支持三种消费模式（轮询 / 广播唤醒 / Stream 消费组），内置任务保活、废弃任务自动重入队、队列背压、节点监控与优雅停机能力。适用于中小规模、无需引入专业消息中间件的分布式任务调度场景。

## 设计初衷与定位

**为什么不用 RabbitMQ / Kafka？**

本项目面向**轻量级项目**而设计。RabbitMQ、Kafka 等专业消息中间件功能完善，但运行成本较高：通常需要 2GB+ 内存以及独立的部署、监控、备份、升级体系，在便宜的云服务器（1~2 核 / 1~2GB 内存）上难以承载——对轻量项目来说，"再维护一个中间件"本身就是最大的成本。

本项目的核心思路是：**复用项目中已有的 Redis（通常用于缓存 / Session），在不引入任何新组件、不增加运维成本的前提下，提供一套够用的分布式任务队列能力**。

这不是"退而求其次"，而是轻量场景下的务实选择：对资源受限、团队规模小、追求稳定运行的项目来说，少一个需要专门伺候的中间件，就少一类运维事故。

**一句话定位**：面向轻量级项目的、尽力而为（best-effort）的分布式任务队列。

## 适用场景与边界

**适合使用的场景**：

- 任务量在每秒几十到几千的量级
- 允许极低概率的消息丢失，或业务上具备补偿手段（重新触发、定时对账）
- 秒级延迟可接受（异步任务、削峰、解耦场景）
- 任务可重试、可幂等设计

**不适合的场景**（请改用专业消息中间件）：

- 丢一条消息就是事故且无法补偿（资金、订单核心链路）
- 每秒万级以上的吞吐量
- 需要延迟队列、严格顺序、事务消息等高级特性

**可靠性语义**：

| 模式 | 投递语义 |
|------|---------|
| Loop / Channel | 尽力投递（best-effort）：存在极小的消息丢失窗口（如节点取出消息后、写入监控 Key 前崩溃），由保活监控兜底重入队 |
| Stream | 接近"至少一次"（at-least-once）：消费组 + ACK 确认 + Pending 恢复，推荐生产使用 |

**消费语义（重要）**：三种模式全部为**点对点（争抢式）**——一条任务只会被**一个**节点取出并执行，多个节点自动分摊负载。Channel 模式名称中的"广播"指的是**唤醒信号**的广播（Pub/Sub 通知所有节点"有活了"），**不是**任务数据的广播；如果需要"每条消息所有节点都处理"的广播消费，本项目不适用（可基于 Stream 多消费组自行扩展）。

## 模块结构

| 模块 | 说明 |
|------|------|
| `queue-execute-core` | 核心实现：执行器、作业基类、任务下发工具、监控与工具类 |
| `queue-execute-starter` | Spring Boot 自动配置：读取 `queue-execute` 配置并创建、启动执行器 |
| `queue-execute-demo` | 使用示例：三种模式的完整演示（含 Web 接口下发任务） |

## 快速开始

### 1. 引入依赖

```xml
<dependency>
    <groupId>com.itdct.quexor</groupId>
    <artifactId>queue-execute-starter</artifactId>
    <version>1.5.0-SNAPSHOT</version>
</dependency>
```

### 2. 添加配置

```yaml
queue-execute:
  enable: true
  queues:
    # 模块名（会被转换为大写作为队列标识）
    loop-queue:
      # 0=轮询 1=广播唤醒 2=Stream消费组
      getJobListenMode: 0
      # 可选：手动指定节点IP，不配置则自动探测
      nodeIp: 192.168.1.102
    stream-queue:
      getJobListenMode: 2
      getJobConfig:
        # Stream消费组名
        streamGroup: default
```

### 3. 定义作业

继承 `BaseQueueJob`，实现核心回调方法：

```java
public class MyJob extends BaseQueueJob {
    @Override
    public void onInit() { /* execute和onAbort前都会调用，用于参数初始化 */ }

    @Override
    public boolean execute() {
        // 核心业务逻辑，返回true表示成功
        return true;
    }

    @Override
    public void onSuccess() { /* 执行成功回调 */ }

    @Override
    public void onAbort() { /* 任务被判定废弃时的回调 */ }

    @Override
    public void onFail() { /* 执行失败回调 */ }

    @Override
    public void onFinish() { /* 无论成功/失败/出错都会执行 */ }

    @Override
    public void onError(Exception e) { /* 执行抛异常时回调 */ }
}
```

作业实例由执行端通过反射创建（无参构造），因此作业类必须提供无参构造函数。

### 4. 下发任务

```java
QueueJobConfig jobConfig = new QueueJobConfig(taskId, "", MyJob.class.getName(), "业务数据JSON");
// 按消费模式选择对应的下发方法
AssignJob.addLoopQueueJob(redisTemplate, "LOOP-QUEUE", jobConfig);
AssignJob.addChannelQueueJob(redisTemplate, "CHANNEL-QUEUE", jobConfig);
AssignJob.addStreamQueueJob(redisTemplate, "STREAM-QUEUE", "default", jobConfig);
```

## 三种消费模式

| 模式 | 常量 | 实现机制 | 可靠性 | 适用场景 |
|------|------|---------|--------|---------|
| 轮询 | `LOOP_GET(0)` | Redis List + 动态间隔轮询 | 中，依赖保活监控兜底 | 简单场景、任务量稳定 |
| 广播唤醒 | `CHANNEL_CAST(1)` | Pub/Sub 唤醒 + List 取数据 | 中，信号可能丢失（超时兜底） | 低延迟要求的突发任务 |
| Stream | `STREAM(2)` | Redis Stream 消费组 + Pending 扫描 | 高，ACK 确认 + pending 恢复 | 要求消息不丢的任务 |

三种模式通用机制：

- **保活刷新**：任务执行期间由独立线程定期刷新 `REFRESH_KEY`，监控线程据此判定任务是否"废弃"（节点崩溃、线程卡死等）
- **废弃重入队**：废弃任务触发 `onAbort` 回调后重新进入队列，`abortTimes` 递增
- **失败归还**：无法实例化等原因无法处理的任务会归还队列（`returnTimes` 递增），超过 `maxReturnTimes` 后进入死信队列
- **任务结果记录**：任务完成时记录结果（成功/失败/出错）与异常摘要（默认保留 7 天，最近 1000 条可查），支持任务追踪
- **死信队列**：超过最大归还次数的任务进入死信队列（默认不自动清理），可通过 `DeadJobManager` 查询、重试、删除
- **幂等扩展点**：重写 `BaseQueueJob.isAlreadyProcessed()` 返回 `true` 可跳过已处理过的任务

## 演示台（queue-execute-demo）

demo 模块内置可视化演示台（端口为 application.yaml 配置的 4000~5000 随机端口）：

- **队列演示** `http://localhost:{port}/queue-ui`：
  - 实时展示三种队列（Loop/Channel/Stream）的状态、等待队列长度、活跃线程与执行统计
  - 选择任务类型（小/中/大）与数量后一键发送，自动追踪消费进度（进度条 + 实时速率曲线）
  - 全部完成后自动计算吞吐量（个/秒）与总耗时
- **任务追踪** `http://localhost:{port}/task-ui`：
  - 查询最近执行完成的任务：结果（成功/失败/出错）、异常摘要、执行节点、端到端耗时、重试次数
- **死信管理** `http://localhost:{port}/dead-ui`：
  - 查看进入死信队列的任务（超过最大归还次数被作废），支持手动重试（重新入队）与删除
- **节点信息** `http://localhost:{port}/node-ui`（根路径 `http://localhost:{port}/` 默认即打开此页）：
  - 按模块分组展示 Redis 中所有活跃节点（节点停止上报后约 5 秒自动从列表消失）
  - 展示每个节点的状态、活跃线程、等待队列、执行统计、CPU/堆内存、最后刷新时间等运行指标

```bash
# 启动演示，需要可用的 Redis 连接（见 application.yaml）
# 1. 首次运行（或修改了 core/starter 后）先安装依赖模块到本地仓库
mvn install -DskipTests
# 2. 进入 demo 目录启动应用
cd queue-execute-demo
mvn spring-boot:run
```

## 典型用法：日志批量入库

把应用日志异步、批量地写入数据库，是轻量项目最典型的使用场景之一。demo 模块内置了完整可运行的示例（`com.itdct.quexe.demo.log` + `LogPersistJob`），链路如下：

```
应用日志 (Logback Appender 采集)
   ↓
内存有界缓冲（默认 5 万条，满则丢弃）
   ↓ 每 200 条 或 每 5 秒（双条件）
攒批器组装“日志批次任务” → AssignJob 下发（LOG-QUEUE 模块，带背压保护）
   ↓
LogPersistJob：解包批次 → JDBC 批量插入
   ↓ 失败：快速重试 → 重新入队（复用归还/死信链路）→ 超过重试上限进死信队列
```

**核心设计**：

- 一条“任务”承载一批日志（默认 200 条），把队列开销摊薄到 1/200；不要一条日志一个任务
- 内存缓冲有界，满了直接丢弃（日志允许少量丢失），不拖垮应用内存
- 数据库不可用时：批次级重试（`BaseQueueJob.retry()` 重新入队），超过 `maxReturnTimes` 进死信队列，可在死信管理页查看与手动重试
- 队列背压（`maxCapacity`）防止 Redis 内存膨胀；入队失败直接丢弃批次，不阻塞业务线程
- 采集端排除采集链路自身的 logger，避免“失败日志再入库”的自我循环

**体验方式**（demo 内置 H2 内存库，无需外部数据库）：

1. 启动 demo 后正常产生日志（框架自身日志即会被采集）
2. 打开任务追踪页 `/task-ui` 选择 LOG-QUEUE：每批日志对应一个任务
3. 打开 H2 控制台 `http://localhost:4050/h2-console`（JDBC URL 填 `jdbc:h2:mem:queuedb`，用户名 `sa`，密码留空）执行 `SELECT COUNT(*) FROM app_log;` 验证入库

**关键配置**：

| 配置 | 默认值 | 说明 |
|------|--------|------|
| `demo.log-batch.batch-size` | 200 | 批次大小（条） |
| `demo.log-batch.flush-interval-millis` | 5000 | 攒批最长等待（毫秒），低流量兜底发送 |
| `demo.log-batch.buffer-capacity` | 50000 | 内存缓冲容量（条），超出丢弃 |
| `queue-execute.queues.log-queue.maxCapacity` | 50000 | 队列背压阈值，防止 Redis 膨胀 |

**注意**：

- 队列作业由框架反射创建，不能直接使用 `@Autowired`；demo 通过静态桥 `SpringBeanHolder` 获取 `JdbcTemplate`（正式项目可采用同样方式）
- 生产使用时为日志表设置清理/归档策略（如按月分区表、定期删除历史数据），避免表无限增长

## 配置参考

### 队列级配置（QueueExeConfig）

| 配置项 | 默认值 | 说明 |
|--------|--------|------|
| `enable` | true | 是否启用该队列 |
| `autoRun` | true | 是否随 Spring 容器自动启动 |
| `getJobListenMode` | 0 | 消费模式：0/1/2 |
| `coreThreadCount` | CPU-1 | 核心执行线程数（最小为1） |
| `maxThreadCount` | CPU*1.5 | 最大执行线程数（最小为1） |
| `queueExecutorQueueSize` | 20 | 执行线程池阻塞队列长度 |
| `threadKeepAliveTime` | 60 | 空闲线程存活时间（秒） |
| `refreshIntervalTime` | 5000 | 任务保活刷新间隔（毫秒） |
| `jobAbortTime` | 40000 | 判定任务废弃的超时时间（毫秒） |
| `initWaitTime` | 1000 | 启动前等待时间（毫秒） |
| `maxCapacity` | 10000000 | 队列最大长度（背压阈值，<=0 表示不限制） |
| `throwErrorIfOverLimit` | true | 超出最大长度时抛异常；false 则循环等待 |
| `overLimitSleepTime` | 1000 | 超出最大长度后的等待时间（毫秒） |
| `canJobExpire` | false | 是否启用任务过期 |
| `totalJobExpireTime` | 600 | 整体任务过期时间（秒） |
| `maxReturnTimes` | 50 | 任务最大归还次数 |
| `notFoundClassReturnQueue` | true | 无法实例化时是否归还队列 |
| `threadPoolShutdownWaitTime` | 60 | 停机时线程池等待时间（秒） |
| `jobClassPackagePrefixes` | 空 | 作业类包名白名单，为空不校验（建议配置以防御反序列化攻击） |
| `recordJobResult` | true | 是否记录任务执行结果（用于任务追踪查询） |
| `jobResultExpireTime` | 604800 | 任务执行结果保留时间（秒，默认 7 天） |
| `jobResultIndexMaxSize` | 1000 | 最近任务索引保留的最大条数（超出后裁剪最旧索引） |

### 子配置

- `getJobConfig`（获取任务）：
  - `getJobErrorWaitTime` 获取出错等待（2000ms）
  - `getJobSleepIntervalTime` 广播模式空转等待（1000ms）
  - `maxNotGetMessageWaitTimes` 广播模式进入休眠前空轮次数（5）
  - `channelAwaitTimeout` 广播模式最大休眠时间（30000ms，防永久阻塞）
  - `streamGroup` Stream 消费组名（default）
  - `streamPendingIntervalTime` pending 扫描间隔（2000ms）
  - `streamPendingTimeout` pending 超时重入队时间（30000ms）
- `monitorJobConfig`（任务监控）：监控间隔（6000ms）、误差偏移（2000ms）、monitorKey TTL（30s）
- `nodeMonitorConfig`（节点上报）：`nodeMonitor` 开关、上报间隔（2000ms）、信息过期时间（5000ms）
- `logConfig`（日志）：完成日志、开始日志、性能日志开关
- `broadcastConfig`（广播）：`broadcastIntervalTime` 发送端节流间隔（1000ms，按模块生效）、`signalIntervalTime` 消费端唤醒防抖（2000ms）

## 运行机制

### 启动与优雅停机

- 执行器由自动配置在**容器就绪阶段**（`SmartLifecycle.start`）统一启动；单个执行器启动异常不影响其他执行器
- 应用未引入 Redis 或未提供 `RedisConnectionFactory` 时，自动配置整体跳过，不影响应用启动；`enable=false` 或未配置 `queues` 时会跳过并输出日志
- 停机流程：置停止标志 → 停止外部监听容器（广播/Stream）→ 等待获取任务/监控/节点信息等线程退出 → 关闭线程池 → 清理节点信息
- 停机后 `shutdownFlag` 保持为 `true`，重新启动必须调用 `execute()`（内部会重置标志），请勿绕过

### 手动创建执行器

不依赖自动配置时，需要手动注册配置并显式启动：

```java
@Bean(destroyMethod = "shutdown")
public StreamQueueExecute streamExecute(RedisTemplate<String, String> redisTemplate) {
    QueueExeConfig config = new QueueExeConfig();
    config.setModuleName("stream-test");
    config.setGetJobListenMode(QueueListenMode.STREAM);

    // 注册配置：使任务下发时能应用背压与广播节流配置
    AssignJob.registerQueueConfig(config.getModuleName(), config);

    StreamQueueExecute executor = new StreamQueueExecute(config, redisTemplate);
    // 手动创建的执行器不会自动启动，需要显式调用
    executor.autoStartIfNeeded();
    return executor;
}
```

### 队列容量与背压

- 任务下发前会检查队列长度，超过 `maxCapacity` 时：
  - `throwErrorIfOverLimit=true`（默认）：抛出 `QueueExecuteException`，由调用方决定如何处理
  - `throwErrorIfOverLimit=false`：循环等待（每轮 `overLimitSleepTime` 毫秒）直到队列低于阈值
- 手动创建执行器或跨应用下发任务时，需要先调用 `AssignJob.registerQueueConfig()` 注册配置，否则不启用容量检查

## 生产使用建议

在保持轻量的前提下，按以下顺序加固即可（成本低、收益高）：

1. **优先使用 Stream 模式**：Redis Stream 是官方为消息队列场景设计的结构（消费组 + ACK + Pending），把 Stream 模式用扎实，可靠性已接近专业 MQ 的投递语义；Loop / Channel 适合对可靠性要求更低的场景
2. **开启 Redis AOF 持久化**（`appendonly yes`）：Redis 进程崩溃重启后队列数据不丢失，这是性价比最高的一道保险，每秒刷盘即可
3. **业务侧做幂等 + 兜底对账**：重写 `BaseQueueJob.isAlreadyProcessed()` 实现幂等；增加定时任务扫描"应完成未完成"的数据做补偿——比追求 MQ 层面的绝对可靠便宜得多
4. **统一团队认知**：明确本项目是"尽力投递、允许补偿"语义，不要假设它等价于专业消息中间件

**何时考虑迁移到专业中间件**：

- Redis 的队列操作成为瓶颈（延迟升高、CPU 打满）
- 业务可靠性要求升级为"绝不能丢"且无法业务补偿
- 出现因丢失消息导致的真实业务损失

## 注意事项

1. **作业类白名单**：生产环境建议配置 `jobClassPackagePrefixes`，防止反序列化攻击
2. **节点ID**：同一队列的节点ID不能重复，未配置时自动生成 `node-{序号}`；如需固定请确保唯一
3. **广播模式可靠性**：Pub/Sub 信号可能丢失，消费端有超时兜底（`channelAwaitTimeout`），对延迟敏感且要求必达的场景请使用 Stream 模式
4. **Stream 模式内存**：长时间运行建议按需对 Stream 执行 `XTRIM` 控制内存增长
5. **JDK 兼容**：Lombok 需 1.18.30+ 以兼容高版本 JDK 编译
6. **配置校验**：`getJobListenMode` 只接受 0/1/2，非法值会导致应用启动失败（避免静默降级为轮询模式）