# 【RuoYi-Eggjs】：基于 Bull Queue 的企业级定时任务调度系统

> 本文介绍 RuoYi-Eggjs 中基于 Bull Queue 构建的企业级定时任务调度系统，涵盖动态任务管理、分布式执行、失败重试、日志监控等核心功能的设计与实现。

## 一、引言：为什么需要强大的定时任务系统？

在企业级应用中，定时任务是不可或缺的基础设施：

- **数据统计报表**：每日/每周/每月的业务数据汇总
- **系统维护**：日志清理、缓存刷新、数据备份
- **业务提醒**：订单超时处理、会员到期通知
- **数据同步**：不同系统间的数据同步任务

RuoYi-Eggjs 基于 **[Bull Queue](https://github.com/OptimalBits/bull)** 构建了一套完整的定时任务调度系统，具备以下企业级特性：

- 🔄 **动态任务管理** - 从数据库读取 cron 表达式，支持热更新
- 🚀 **分布式执行** - 基于 Redis 的分布式任务调度
- 🔁 **失败重试** - 自动重试机制，保证任务执行可靠性
- 📊 **可视化监控** - 完整的任务执行日志和状态监控
- ⚡ **手动执行** - 支持立即执行任务，方便测试调试

## 二、系统架构设计

### 整体架构

```
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   管理端 Web     │    │   Bull Queue    │    │   任务执行器     │
│                 │    │                 │    │                 │
│ ┌─────────────┐ │    │ ┌─────────────┐ │    │ ┌─────────────┐ │
│ │ 任务管理API  │ │───▶│ │ Redis 队列   │ │───▶│ │ ryTask 处理器│ │
│ │ - 增删改查   │ │    │ │ - 任务调度   │ │    │ │ - 任务执行   │ │
│ │ - 启停控制   │ │    │ │ - 失败重试   │ │    │ │ - 结果记录   │ │
│ └─────────────┘ │    │ └─────────────┘ │    │ └─────────────┘ │
└─────────────────┘    └─────────────────┘    └─────────────────┘
         │                       │                       │
         ▼                       ▼                       ▼
┌─────────────────────────────────────────────────────────────────┐
│                     数据存储层                                   │
│  ┌─────────────────┐            ┌─────────────────────────────┐ │
│  │     MySQL       │            │          Redis              │ │
│  │ - sys_job 表    │            │ - 队列数据                   │ │
│  │ - sys_job_log表 │            │ - 分布式锁                   │ │
│  └─────────────────┘            └─────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```

### 核心组件

| 组件 | 职责 | 实现文件 |
|------|------|----------|
| **JobService** | 任务管理、调度控制 | `app/service/monitor/job.js` |
| **Bull Queue** | 分布式任务队列 | `app/queue/ryTask.js` |
| **CronUtils** | Cron 表达式工具 | `app/util/cronUtils.js` |
| **JobLog** | 任务执行日志 | `app/service/monitor/jobLog.js` |
| **RyTask** | 任务执行器 | `app/service/ryTask.js` |

## 三、Cron 表达式处理

### 智能 Cron 工具类

系统内置了强大的 Cron 表达式处理工具，支持表达式验证、下次执行时间计算等功能：

```js
// app/util/cronUtils.js
const parser = require('cron-parser');

class CronUtils {
  // 验证 cron 表达式是否有效
  static isValid(cronExpression) {
    try {
      parser.parseExpression(cronExpression);
      return true;
    } catch (err) {
      return false;
    }
  }

  // 获取下次执行时间
  static getNextExecution(cronExpression) {
    try {
      const interval = parser.parseExpression(cronExpression);
      return interval.next().toDate();
    } catch (err) {
      return null;
    }
  }
}
```

### 常用 Cron 表达式

系统预置了常用的 Cron 表达式模板：

```js
static get EXAMPLES() {
  return {
    EVERY_MINUTE: '0 * * * * *',           // 每分钟执行
    EVERY_HOUR: '0 0 * * * *',             // 每小时执行
    EVERY_DAY: '0 0 0 * * *',              // 每天 0 点执行
    WORKDAY_MORNING: '0 0 9 * * 1-5',      // 工作日早上 9 点
    EVERY_WEEK: '0 0 0 * * 0',             // 每周日 0 点执行
  };
}
```

## 四、Bull Queue 队列处理

### 队列初始化

基于 Redis 创建分布式任务队列：

```js
// app/queue/ryTask.js
const Queue = require('bull');

module.exports = app => {
  const queue = new Queue('ryTask', {
    redis: {
      port: app.config.bull.client.port,
      host: app.config.bull.client.host,
      password: app.config.bull.client.password,
      db: app.config.bull.client.db,
    },
  });

  // 配置任务处理器
  queue.process(async (job) => {
    const ctx = app.createAnonymousContext();
    const { invokeTarget, jobInfo } = job.data;
    
    // 执行任务逻辑...
  });

  return queue;
};
```

### 任务执行流程

每个任务的执行都经过完整的生命周期管理：

```js
queue.process(async (job) => {
  const startTime = Date.now();
  let status = '0'; // 0-成功 1-失败
  let jobMessage = '';
  let exceptionInfo = '';

  try {
    // 1. 记录开始执行
    ctx.logger.info(`[Bull] 开始执行任务: ${jobInfo.jobName}`);
    
    // 2. 调用任务执行器
    const result = await ctx.service.ryTask.execute(invokeTarget);
    
    jobMessage = result.message || '任务执行成功';
    ctx.logger.info(`[Bull] 任务执行成功: ${jobInfo.jobName}`);
  } catch (err) {
    status = '1';
    jobMessage = '任务执行失败';
    exceptionInfo = err.message;
    
    ctx.logger.error(`[Bull] 任务执行失败: ${jobInfo.jobName}`, err);
    throw err; // 抛出错误，让 Bull 处理重试
  } finally {
    const duration = Date.now() - startTime;
    
    // 3. 记录执行日志
    await ctx.service.monitor.jobLog.insertJobLog({
      jobName: jobInfo.jobName,
      jobGroup: jobInfo.jobGroup,
      invokeTarget: invokeTarget,
      jobMessage: `${jobMessage} (耗时: ${duration}ms)`,
      status,
      exceptionInfo: exceptionInfo.substring(0, 2000),
      createTime: ctx.helper.formatDate(new Date(startTime)),
    });
  }
});
```

## 五、动态任务管理

### 创建定时任务

系统支持动态创建定时任务，任务信息存储在数据库中：

```js
// app/service/monitor/job.js
async createBullJob(job) {
  const { app, ctx } = this;
  
  try {
    // 使用 jobId + invokeTarget 作为唯一标识
    const uniqueId = `${job.jobId}:${job.invokeTarget}`;
    
    // 清理可能存在的旧任务
    await this.cleanOldRepeatableJobs(job);
    
    // 添加新的重复任务
    await app.queue.ryTask.add(
      {
        invokeTarget: job.invokeTarget,
        jobInfo: {
          jobId: job.jobId,
          jobName: job.jobName,
          jobGroup: job.jobGroup,
          uniqueId,
        },
      },
      {
        jobId: uniqueId,
        repeat: {
          cron: job.cronExpression,
          key: uniqueId,
        },
        removeOnComplete: true,
        removeOnFail: 100,
      }
    );
    
    ctx.logger.info(`[Bull] 创建定时任务成功: ${job.jobName}`);
    return true;
  } catch (err) {
    ctx.logger.error(`[Bull] 创建定时任务失败: ${job.jobName}`, err);
    return false;
  }
}
```

### 任务状态控制

支持任务的启动、暂停、删除等状态控制：

```js
// 暂停任务
async pauseBullJob(job) {
  return await this.deleteBullJob(job);
}

// 恢复任务
async resumeBullJob(job) {
  return await this.createBullJob(job);
}

// 立即执行任务
async runBullJob(job) {
  await app.queue.ryTask.add(
    {
      invokeTarget: job.invokeTarget,
      jobInfo: job,
    },
    {
      removeOnComplete: true,
      priority: 1, // 高优先级立即执行
    }
  );
  
  return true;
}
```

## 六、任务执行与监控

### 任务执行日志

系统完整记录每次任务执行的详细日志：

```js
// app/service/monitor/jobLog.js
async insertJobLog(jobLog) {
  const { ctx } = this;
  
  try {
    const mapper = ctx.helper.getDB(ctx).sysJobLogMapper;
    
    // 设置日志ID和创建时间
    jobLog.jobLogId = ctx.helper.generateId();
    jobLog.createTime = jobLog.createTime || ctx.helper.formatDate(new Date());
    
    return await mapper.insertJobLog([], jobLog);
  } catch (err) {
    ctx.logger.error('插入任务执行日志失败:', err);
    throw err;
  }
}
```

### 日志查询与统计

提供丰富的日志查询功能，支持按任务名、执行状态、时间范围等条件筛选：

```js
async selectJobLogPage(params = {}) {
  const { ctx } = this;
  const mapper = ctx.helper.getDB(ctx).sysJobLogMapper;

  return await ctx.helper.pageQuery(
    mapper.selectJobLogListMapper([], params),
    params,
    mapper.db()
  );
}
```

## 七、企业级特性

### 1. 分布式支持

基于 Redis 实现分布式任务调度，支持多实例部署：

- **任务分发**：Redis 作为消息队列，确保任务在集群中合理分发
- **状态同步**：任务状态和执行结果通过 Redis 在各实例间同步
- **防重复执行**：通过 Redis 锁机制防止同一任务重复执行

### 2. 失败重试机制

Bull Queue 内置强大的失败重试功能：

```js
// 配置重试策略
{
  attempts: 3,                    // 重试次数
  backoff: {
    type: 'exponential',          // 指数退避
    delay: 2000,                  // 初始延迟
  },
  removeOnFail: 100,              // 保留失败任务数量
}
```

### 3. 性能监控

系统提供完整的性能监控指标：

- **执行时长统计**：记录每次任务执行的详细耗时
- **成功/失败率**：统计任务执行的成功率
- **队列状态监控**：监控队列积压情况

### 4. 安全控制

严格的任务执行安全控制：

```js
// 任务安全校验
checkJobSecurity(job) {
  const securityCheck = { pass: true, message: '' };
  
  // 检查调用目标是否安全
  if (job.invokeTarget && !job.invokeTarget.match(/^[\w\.]+($.*$)?$/)) {
    securityCheck.pass = false;
    securityCheck.message = '调用目标格式不正确';
  }
  
  return securityCheck;
}
```

## 八、最佳实践

### 1. 任务设计原则

- **幂等性**：确保任务可以安全重复执行
- **异常处理**：完善的错误处理和日志记录
- **资源管控**：避免长时间占用系统资源

### 2. Cron 表达式优化

```js
// 推荐的 Cron 表达式
'0 0 2 * * *'          // 每天凌晨2点（避开高峰期）
'0 0 9-17 * * 1-5'     // 工作日工作时间
'0 */5 * * * *'        // 每5分钟（适度频率）

// 避免的表达式
'* * * * * *'          // 每秒执行（过于频繁）
'0 0 0 * * *'          // 午夜12点（高峰期）
```

### 3. 监控告警

建议配置任务执行监控告警：

```js
// 监控失败任务
queue.on('failed', (job, err) => {
  // 发送告警通知
  notificationService.sendAlert({
    title: '任务执行失败',
    content: `任务 ${job.data.jobInfo.jobName} 执行失败: ${err.message}`,
    level: 'error'
  });
});
```

## 九、总结

RuoYi-Eggjs 的定时任务调度系统通过以下技术栈实现了企业级的任务管理能力：

### 核心优势

1. **技术先进**：基于 Bull Queue + Redis 的成熟方案
2. **功能完整**：涵盖任务的全生命周期管理
3. **高可用性**：分布式部署 + 失败重试保证系统稳定
4. **易于使用**：直观的管理界面 + 丰富的 API


-----


+ 项目地址：[RuoYi-Eggjs](https://github.com/undsky/ruoyi-eggjs)
+ 开发文档：[RuoYi-Eggjs 文档](https://www.undsky.com/blog/?category=RuoYi-Eggjs#)

本文介绍 RuoYi-Eggjs 中基于 Bull Queue 构建的企业级定时任务调度系统，涵盖动态任务管理、分布式执行、失败重试、日志监控等核心功能的设计与实现。

【RuoYi-Eggjs】：基于 Bull Queue 的企业级定时任务调度系统

编程语言

前端

RuoYi-Eggjs基于BullQueue构建企业级定时任务调度系统，支持动态任务管理、分布式执行、失败重试和可视化监控。提供Cron表达式处理、任务执行日志、安全控制等功能，适用于数据统计、系统维护等场景。采用Redis实现分布式调度，确保高可用性和稳定性，是企业应用的理想定时任务解决方案。

消息队列

数据备份

数据存储

数据同步

数据库

MySQL

Redis

腾讯云数据库特惠，新用户首单0.5折起


tencentdb-catalog

datainlong

message-queue-catalog

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【RuoYi-Eggjs】：基于 Bull Queue 的企业级定时任务调度系统-腾讯云开发者社区-腾讯云

【RuoYi-Eggjs】：基于 Bull Queue 的企业级定时任务调度系统

【RuoYi-Eggjs】：基于 Bull Queue 的企业级定时任务调度系统

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐