系统队列任务文档说明.md 14 KB
CRMEB 系统队列任务文档说明
概述
CRMEB 系统采用基于 Redis 的消息队列机制，通过 ThinkPHP 框架的队列组件实现异步任务处理。队列系统在提高系统性能、优化用户体验、保证数据一致性方面发挥着重要作用。
队列系统架构
1. 技术栈
- 队列驱动: Redis (主) + Database (备) + Sync (同步)
- 队列服务: ThinkPHP Queue
- 定时任务: Workerman + Crontab
- 任务调度: 基于事件的异步处理
2. 核心组件
2.1 队列配置 (config/queue.php)
return [
'default' => 'redis', // 默认驱动
'prefix' => 'crmeb_', // 队列前缀
'connections' => [
'redis' => [
'driver' => 'redis',
'queue' => 'CRMEB', // 队列名称
'host' => '127.0.0.1',
'port' => 6379,
'password' => '',
'select' => 0,
],
],
'failed' => [
'type' => 'database',
'table' => 'failed_jobs', // 失败任务表
],
];
2.2 基础任务类 (BaseJobs)
abstract class BaseJobs implements JobInterface
{
// 任务执行入口
public function fire(Job $job, $data): void
// 任务重试机制
protected function runJob(string $action, Job $job, array $infoData, int $errorCount = 3)
}
2.3 队列特性 (QueueTrait)
trait QueueTrait
{
// 立即推送任务
public static function dispatch($action, array $data = [], string $queueName = null)
// 延迟推送任务
public static function dispatchSecs(int $secs, $action, array $data = [], string $queueName = null)
}
队列任务分类
1. 订单相关任务
1.1 OrderJob - 订单处理任务
功能: 订单支付成功后的后续处理 触发时机: 订单支付成功时 主要处理:
- 计算商品节省金额
- 更新用户支付订单数量
- 检测并设置推广人身份
- 添加用户购买标签
- 发送客服通知消息
- 检测会员等级升级
- 向后台发送新订单消息
// 使用示例
OrderJob::dispatch('doJob', [$orderInfo]);
1.2 OrderCreateAfterJob - 订单创建后处理
功能: 订单创建后的数据处理 触发时机: 订单创建完成后 主要处理:
- 计算订单实际价格
- 处理推广关系
- 计算佣金分配
- 更新订单附加信息
1.3 UnpaidOrderCancelJob - 未支付订单取消
功能: 自动取消超时未支付订单 触发时机: 定时任务或延迟队列 处理逻辑:
- 检查订单支付状态
- 回退积分和优惠券
- 恢复商品库存
- 标记订单为已取消
// 延迟30分钟后执行
UnpaidOrderCancelJob::dispatchSecs(1800, 'doJob', [$orderId]);
2. 商品相关任务
2.1 ProductLogJob - 商品日志记录
功能: 记录商品相关行为日志 触发时机: 商品浏览、加购、下单等 日志类型:
visit: 商品访问cart: 加入购物车order: 下单pay: 支付collect: 收藏refund: 退款
ProductLogJob::dispatch('doJob', ['visit', $productData]);
2.2 ProductStockJob - 商品库存处理
功能: 商品库存计算和更新 触发时机: 商品操作后 处理功能:
- 分布式库存计算
- 规格库存汇总
- 库存同步更新
2.3 ProductCopyJob - 商品复制任务
功能: 异步复制商品信息 触发时机: 商品复制操作 处理内容:
- 商品基本信息复制
- 规格属性复制
- 图片资源复制
3. 通知相关任务
3.1 SmsJob - 短信发送任务
功能: 异步发送短信通知 触发时机: 各种业务事件 支持场景:
- 订单通知
- 支付提醒
- 注册验证
- 营销推广
SmsJob::dispatch('doJob', [$phone, $data, $template]);
3.2 TemplateJob - 模板消息任务
功能: 发送微信模板消息 触发时机: 业务状态变更 消息类型:
- 订单状态通知
- 支付成功通知
- 物流更新通知
3.3 SyncMessageJob - 消息同步任务
功能: 多端消息同步 触发时机: 消息发送时 同步范围:
- 客服系统
- 管理后台
- 用户端
4. 营销相关任务
4.1 PinkJob - 拼团任务
功能: 拼团活动处理 触发时机: 拼团相关操作 处理内容:
- 拼团状态更新
- 拼团失败处理
- 拼团成功通知
4.2 LiveJob - 直播任务
功能: 直播相关处理 触发时机: 直播事件 处理内容:
- 直播状态同步
- 商品上下架
- 观众数据统计
4.3 AutoCommentJob - 自动评论任务
功能: 自动生成商品评论 触发时机: 订单完成后 处理逻辑:
- 根据商品信息生成评论
- 模拟用户评价
- 提升商品活跃度
5. 系统维护任务
5.1 TaskJob - 定时任务
功能: 系统定时维护 执行频率: 每日定时 维护内容:
- 清理过期文件
- 数据统计汇总
- 系统健康检查
// 清理昨日海报
TaskJob::dispatch('emptyYesterdayAttachment');
5.2 UpgradeJob - 系统升级任务
功能: 系统升级处理 触发时机: 手动触发或定时 处理内容:
- 数据库结构更新
- 配置文件升级
- 兼容性检查
5.3 CheckQueueJob - 队列检查任务
功能: 监控队列健康状态 执行频率: 定时检查 检查项目:
- 队列堆积情况
- 失败任务统计
- 系统资源使用
6. 用户相关任务
6.1 UserJob - 用户处理任务
功能: 用户相关异步处理 触发时机: 用户操作 处理内容:
- 用户等级计算
- 积分统计分析
- 行为数据更新
6.2 AgentJob - 代理任务
功能: 代理商相关处理 触发时机: 代理业务 处理内容:
- 佣金计算
- 等级升级
- 数据统计
7. 物流相关任务
7.1 OrderExpressJob - 订单物流任务
功能: 物流信息处理 触发时机: 物流状态变更 处理内容:
- 物流信息同步
- 状态更新通知
- 异常情况处理
7.2 TakeOrderJob - 订单接收任务
功能: 订单接收处理 触发时机: 订单创建 处理内容:
- 订单数据验证
- 库存检查
- 自动分配
8. 财务相关任务
8.1 OrderInvoiceJob - 订单发票任务
功能: 发票处理 触发时机: 发票申请 处理内容:
- 发票生成
- 状态同步
- 通知发送
8.2 OutPushJob - 外部推送任务
功能: 数据外部推送 触发时机: 数据变更 推送内容:
- 订单数据
- 用户数据
- 商品数据
定时任务系统
1. Timer 命令
基于 Workerman 实现的定时任务调度器
# 启动定时任务
php think timer start
# 守护进程方式启动
php think timer start -d
# 停止定时任务
php think timer stop
# 重启定时任务
php think timer reload
2. 系统定时任务
数据库表: system_crontab
字段说明:
name: 任务名称task_type: 任务类型mark: 任务标识content: 任务内容max_execution_time: 最大执行时间execution_cycle: 执行周期is_open: 是否开启next_execution_time: 下次执行时间last_execution_time: 上次执行时间
3. 内置定时任务
3.1 订单相关
- 未支付订单取消: 每分钟检查一次
- 预售商品下架: 每小时检查一次
- 拼团失败处理: 每5分钟检查一次
3.2 数据统计
- 日统计汇总: 每日凌晨执行
- 用户行为分析: 每小时执行
- 销售数据统计: 每日执行
3.3 系统维护
- 日志清理: 每周执行
- 临时文件清理: 每日执行
- 缓存刷新: 定时执行
队列监控与管理
1. 队列状态监控
// 检查队列状态
Queue::instance()->getQueueInfo();
// 获取队列长度
Queue::instance()->getQueueLength();
// 获取失败任务
Queue::instance()->getFailedJobs();
2. 任务管理
// 重新执行失败任务
Queue::instance()->retryFailed($jobId);
// 清空队列
Queue::instance()->clearQueue($queueName);
// 暂停队列
Queue::instance()->pauseQueue($queueName);
3. 日志记录
所有队列任务都有详细的日志记录:
- 成功执行日志
- 失败重试日志
- 异常堆栈信息
- 性能执行统计
队列配置最佳实践
1. 队列分组
根据业务重要性分组:
// 高优先级队列
'order_pay' => 'CRMEB_ORDER_PAY'
'sms' => 'CRMEB_SMS'
'email' => 'CRMEB_EMAIL'
// 普通优先级队列
'statistics' => 'CRMEB_STATISTICS'
'log' => 'CRMEB_LOG'
// 低优先级队列
'report' => 'CRMEB_REPORT'
'cleanup' => 'CRMEB_CLEANUP'
2. 重试策略
// 不同任务的重试次数
'order_pay' => 3 // 支付任务重试3次
'sms' => 5 // 短信任务重试5次
'email' => 3 // 邮件任务重试3次
'statistics' => 1 // 统计任务重试1次
3. 延迟设置
// 常见延迟时间
UnpaidOrderCancelJob::dispatchSecs(1800, 'doJob', [$orderId]); // 30分钟
OrderRemindJob::dispatchSecs(86400, 'doJob', [$orderId]); // 24小时
CleanupJob::dispatchSecs(604800, 'doJob', [$data]); // 7天
性能优化建议
1. 队列优化
- 使用 Redis 集群提高可用性
- 合理设置队列数量避免热点
- 定期清理过期队列数据
- 监控队列长度及时扩容
2. 任务优化
- 避免单个任务过于复杂
- 合理拆分大任务为小任务
- 使用事务保证数据一致性
- 添加超时控制避免阻塞
3. 资源优化
- 控制并发任务数量
- 合理设置内存限制
- 优化数据库查询性能
- 使用连接池减少开销
故障处理
1. 常见问题
1.1 队列堆积
现象: 队列长度持续增长 原因: 消费速度小于生产速度 解决:
- 增加消费者进程数
- 优化任务执行效率
- 检查系统资源使用
1.2 任务重复执行
现象: 同一任务被执行多次 原因: 任务重复入队或幂等性问题 解决:
- 添加任务唯一标识
- 实现幂等性检查
- 优化任务重试逻辑
1.3 内存泄漏
现象: 进程内存持续增长 原因: 任务中存在资源未释放 解决:
- 检查对象引用
- 及时释放大对象
- 定期重启消费者进程
2. 应急处理
2.1 队列暂停
# 暂停所有队列
php think queue:pause
# 暂停指定队列
php think queue:pause --queue=order_pay
2.2 强制清空
# 清空指定队列
php think queue:clear --queue=statistics
# 清空所有队列
php think queue:clear --all
2.3 重建队列
# 重建队列表
php think queue:table
# 重新发布失败任务
php think queue:retry all
开发指南
1. 创建新任务
1.1 创建任务类
<?php
namespace app\jobs;
use crmeb\basic\BaseJobs;
use crmeb\traits\QueueTrait;
class CustomJob extends BaseJobs
{
use QueueTrait;
/**
* 执行任务
* @param array $data
* @return bool
*/
public function doJob(array $data): bool
{
try {
// 业务逻辑处理
$this->processData($data);
return true;
} catch (\Exception $e) {
// 记录错误日志
Log::error('任务执行失败: ' . $e->getMessage());
return false;
}
}
private function processData(array $data): void
{
// 具体业务逻辑
}
}
1.2 调度任务
// 立即执行
CustomJob::dispatch('doJob', [$data]);
// 延迟执行
CustomJob::dispatchSecs(300, 'doJob', [$data]);
// 指定队列
CustomJob::dispatch('doJob', [$data], 'custom_queue');
2. 测试队列任务
2.1 单元测试
public function testCustomJob()
{
$job = new CustomJob();
$result = $job->doJob($testData);
$this->assertTrue($result);
}
2.2 集成测试
public function testQueueDispatch()
{
// 模拟队列调度
Queue::fake();
CustomJob::dispatch('doJob', [$testData]);
Queue::assertPushed(CustomJob::class, function ($job) use ($testData) {
return $job->data[0] === $testData;
});
}
3. 监控和调试
3.1 添加日志
public function doJob(array $data): bool
{
Log::info('任务开始执行', ['data' => $data]);
try {
// 业务逻辑
Log::info('任务执行成功');
return true;
} catch (\Exception $e) {
Log::error('任务执行失败', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
]);
return false;
}
}
3.2 性能监控
public function doJob(array $data): bool
{
$startTime = microtime(true);
try {
// 业务逻辑
$endTime = microtime(true);
$executeTime = ($endTime - $startTime) * 1000; // 毫秒
Log::info('任务执行完成', [
'execute_time' => $executeTime . 'ms',
'memory_usage' => memory_get_usage(true)
]);
return true;
} catch (\Exception $e) {
Log::error('任务执行失败', ['error' => $e->getMessage()]);
return false;
}
}
总结
CRMEB 的队列任务系统是系统高性能运行的关键组件，通过合理的任务分类、优先级管理、错误处理和监控机制，确保了系统的稳定性和可扩展性。在实际使用中，需要根据业务特点合理设计任务粒度，优化执行效率，并建立完善的监控告警机制。
队列系统的正确使用可以显著提升用户体验，降低系统负载，提高数据处理的一致性。建议开发团队充分理解队列机制，在合适场景下合理应用，并持续优化和改进队列性能。
提示：该文档由AI生成，仅供参考。