This commit is contained in:
张成
2025-12-17 17:30:03 +08:00
parent 663a5f7fa6
commit 735fa0c621
6 changed files with 495 additions and 124 deletions

176
_doc/task.md Normal file
View File

@@ -0,0 +1,176 @@
# task_status_${snCode} 主题推送位置分析
## 一、推送位置(服务端)
### 1. 任务状态变更推送 (`task_status_update`)
**位置**: `autoAiWorkSys/api/middleware/schedule/taskQueue.js`
**方法**: `notifyTaskStatusChange(sn_code, taskData)`
**推送时机**:
-**任务开始执行时** (第586行)
- 状态: `running`
- 进度: `0`
- 触发位置: `executeTask` 方法中,任务开始执行时
-**任务完成时** (第631行)
- 状态: `completed`
- 进度: `100`
- 触发位置: `executeTask` 方法中,任务成功完成后
-**任务失败时** (第675行)
- 状态: `failed`
- 进度: `0`
- 错误信息: `errorMessage`
- 触发位置: `executeTask` 方法中,任务执行失败时
**推送内容**:
```javascript
{
action: 'task_status_update',
data: {
taskId: task.id,
taskName: task.taskName,
taskType: task.taskType,
status: 'running' | 'completed' | 'failed',
progress: 0 | 100,
errorMessage?: string,
endTime?: Date
},
timestamp: string
}
```
### 2. 任务状态摘要推送 (`task_status_summary`)
**位置**: `autoAiWorkSys/api/middleware/schedule/taskQueue.js`
**方法**: `sendTaskStatusSummary(sn_code)`
**推送时机**:
-**定时推送** (每10秒一次)
- 位置: `autoAiWorkSys/api/middleware/schedule/scheduledJobs.js`
- 方法: `syncTaskStatusSummary()`
- 条件: 只向在线设备推送(最后心跳时间 < 3分钟
**推送内容**:
```javascript
{
action: 'task_status_summary',
data: {
sn_code: string,
currentTask: {
taskId: number,
taskName: string,
taskType: string,
status: 'running',
progress: number,
currentStep: string,
startTime: Date,
jobTitle?: string, // 新增:职位名称
companyName?: string // 新增:公司名称
} | null,
pendingTasks: Array<{
taskId: number,
taskName: string,
taskType: string,
status: 'pending',
scheduledTime: Date,
priority: number
}>,
nextTaskTime: Date | null,
pendingCount: number, // 队列中的任务数
totalPendingCount: number, // 总待执行数(包括当前任务的剩余步骤)
mqttTopic: string,
timestamp: string
},
timestamp: string
}
```
## 二、接收位置(客户端)
### 1. MQTT 订阅
**位置**: `boss-automation-nodejs/src/services/mqttService.js`
**订阅时机**: MQTT 连接成功后 (第360行)
**订阅代码**:
```javascript
const taskStatusTopic = `task_status_${this.config.snCode}`;
this.client.subscribe(taskStatusTopic, { qos: 1 });
```
### 2. 消息处理
**位置**: `boss-automation-nodejs/src/services/mqttService.js`
**处理方法**: `handleMessage` (第292行)
**处理逻辑**:
- 如果 `action === 'task_status_update'`:
- 发送到渲染进程: `task:status-update`
- 如果 `action === 'task_status_summary'`:
- 发送到渲染进程: `task:status-summary`
### 3. 渲染进程接收
**位置**: `boss-automation-nodejs/app/mixins/taskMixin.js`
**方法**:
- `onTaskStatusUpdate(taskData)` - 处理任务状态变更
- `onTaskStatusSummary(summary)` - 处理任务状态摘要
**事件监听**: `boss-automation-nodejs/app/mixins/eventListenerMixin.js`
## 三、当前推送策略分析
### ✅ 优点
1. **实时性**: 任务状态变更立即推送
2. **完整性**: 定时推送完整的状态摘要
3. **可靠性**: 只向在线设备推送,避免资源浪费
### ⚠️ 潜在问题
1. **推送频率**: 定时任务每10秒推送一次可能过于频繁
2. **重复推送**: 任务状态变更和定时摘要可能推送重复信息
3. **推送时机**: 某些关键状态变更可能没有及时推送(如进度更新)
## 四、建议的优化方案
### 方案1: 减少定时推送频率
- 当前: 每10秒推送一次
- 建议: 改为每30秒或60秒推送一次
- 理由: 任务状态变更已实时推送,定时推送主要用于同步完整状态
### 方案2: 只在关键时机推送摘要
- 任务队列变化时推送(新增/删除任务)
- 当前任务完成/失败时推送
- 定时推送作为兜底(频率降低)
### 方案3: 合并推送
- 将任务状态变更和摘要合并为一个消息
- 减少消息数量,提高效率
### 方案4: 添加进度更新推送
- 任务执行过程中,定期推送进度更新
- 例如每完成10%进度推送一次
## 五、需要推送的场景建议
### ✅ 必须推送的场景
1. **任务开始执行** - 立即推送状态变更
2. **任务完成** - 立即推送状态变更 + 状态摘要
3. **任务失败** - 立即推送状态变更 + 状态摘要
4. **任务队列变化** - 新增/删除任务时推送摘要
### ⚠️ 可选推送的场景
1. **进度更新** - 每10%或每完成一个步骤推送一次
2. **定时同步** - 作为兜底频率降低到30-60秒
### ❌ 不需要推送的场景
1. **任务状态未变化** - 避免重复推送相同状态
2. **设备离线** - 不向离线设备推送

View File

@@ -127,6 +127,26 @@ class CommandManager {
// 4. 更新指令状态为运行中
await this._update_command_status(command_id, 'running', null, null, start_time);
// 4.5 推送指令开始执行状态
try {
const deviceWorkStatusNotifier = require('./deviceWorkStatusNotifier');
const taskQueue = require('./taskQueue');
const summary = await taskQueue.getTaskStatusSummary(task.sn_code);
await deviceWorkStatusNotifier.sendDeviceWorkStatus(task.sn_code, summary, {
currentCommand: {
command_id: command_id,
command_name: command_name,
command_type: command_type,
command_params: command_params,
progress: 0,
startTime: start_time.toISOString()
}
});
} catch (pushError) {
// 推送失败不影响指令执行
console.warn(`[指令管理] 推送设备工作状态失败:`, pushError.message);
}
// 5. 执行指令(统一封装)
const result = await this._execute_command_with_timeout(
command_id,
@@ -141,6 +161,17 @@ class CommandManager {
// 6. 记录成功结果
await this._record_command_result(command_id, 'completed', result, null, start_time);
// 6.5 推送指令完成状态
try {
const deviceWorkStatusNotifier = require('./deviceWorkStatusNotifier');
const taskQueue = require('./taskQueue');
const summary = await taskQueue.getTaskStatusSummary(task.sn_code);
await deviceWorkStatusNotifier.sendDeviceWorkStatus(task.sn_code, summary);
} catch (pushError) {
// 推送失败不影响指令执行
console.warn(`[指令管理] 推送设备工作状态失败:`, pushError.message);
}
return {
command_id: command_id,
command_name: command_name,
@@ -159,6 +190,17 @@ class CommandManager {
error,
start_time
);
// 推送指令失败状态
try {
const deviceWorkStatusNotifier = require('./deviceWorkStatusNotifier');
const taskQueue = require('./taskQueue');
const summary = await taskQueue.getTaskStatusSummary(task.sn_code);
await deviceWorkStatusNotifier.sendDeviceWorkStatus(task.sn_code, summary);
} catch (pushError) {
// 推送失败不影响错误处理
console.warn(`[指令管理] 推送设备工作状态失败:`, pushError.message);
}
}
// 重新抛出错误,让调用方知道执行失败

View File

@@ -0,0 +1,253 @@
/**
* 设备工作状态推送服务
* 负责向客户端推送设备当前工作状态(任务、指令等)
*/
const db = require('../dbProxy');
class DeviceWorkStatusNotifier {
constructor() {
this.mqttClient = null;
}
/**
* 设置 MQTT 客户端
* @param {object} mqttClient - MQTT 客户端实例
*/
setMqttClient(mqttClient) {
this.mqttClient = mqttClient;
}
/**
* 获取 MQTT 客户端
* @returns {Promise<object>} MQTT 客户端实例
*/
async getMqttClient() {
if (this.mqttClient) {
return this.mqttClient;
}
// 从调度系统获取 MQTT 客户端(与 taskQueue 保持一致)
try {
const scheduleManager = require('./index');
if (scheduleManager.mqttClient) {
return scheduleManager.mqttClient;
}
} catch (error) {
// 调度系统未初始化
}
return null;
}
/**
* 发送设备工作状态(统一方法,包含任务和指令状态)
* @param {string} sn_code - 设备SN码
* @param {object} taskStatusSummary - 任务状态摘要(从 taskQueue.getTaskStatusSummary 获取)
* @param {object} options - 可选参数
* @param {object} options.currentCommand - 当前执行的指令信息(可选)
*/
async sendDeviceWorkStatus(sn_code, taskStatusSummary, options = {}) {
try {
const mqttClient = await this.getMqttClient();
if (!mqttClient) {
return; // MQTT客户端不可用静默失败
}
// 构建设备工作状态数据
const workStatus = {
sn_code: taskStatusSummary.sn_code || sn_code,
timestamp: new Date().toISOString(),
// 当前活动(任务或指令)
currentActivity: null,
// 待执行队列
pendingQueue: {
count: taskStatusSummary.pendingCount || 0,
totalCount: taskStatusSummary.totalPendingCount || 0,
nextExecuteTime: taskStatusSummary.nextTaskTime ? new Date(taskStatusSummary.nextTaskTime).toISOString() : null,
tasks: taskStatusSummary.pendingTasks || []
},
// 设备状态(从 deviceManager 获取)
deviceStatus: {
isOnline: true, // 默认在线,实际应从 deviceManager 获取
workMode: 'auto'
}
};
// 如果有当前执行的指令,优先显示指令状态
if (options.currentCommand) {
const cmd = options.currentCommand;
workStatus.currentActivity = {
type: 'command',
id: cmd.command_id || cmd.id,
name: cmd.command_name || cmd.name || '执行指令',
description: this._formatCommandDescription(cmd),
status: 'running',
progress: cmd.progress || 0,
currentStep: cmd.currentStep || '',
startTime: cmd.startTime || new Date().toISOString()
};
}
// 如果有当前执行的任务,显示任务状态
else if (taskStatusSummary.currentTask) {
const task = taskStatusSummary.currentTask;
workStatus.currentActivity = {
type: 'task',
id: task.taskId,
name: task.taskName || task.taskType || '未知任务',
description: this._formatTaskDescription(task),
status: task.status || 'running',
progress: task.progress || 0,
currentStep: task.currentStep || '',
startTime: task.startTime ? new Date(task.startTime).toISOString() : new Date().toISOString()
};
}
// 格式化显示文案(服务端统一处理,客户端直接显示)
workStatus.displayText = this._formatDisplayText(workStatus);
// 格式化下次执行时间显示文案
if (workStatus.pendingQueue.nextExecuteTime) {
const nextTime = new Date(workStatus.pendingQueue.nextExecuteTime);
const now = new Date();
const diff = nextTime - now;
if (diff < 0) {
workStatus.pendingQueue.nextExecuteTimeText = '即将执行';
} else if (diff < 60000) {
workStatus.pendingQueue.nextExecuteTimeText = `${Math.floor(diff / 1000)}秒后`;
} else if (diff < 3600000) {
workStatus.pendingQueue.nextExecuteTimeText = `${Math.floor(diff / 60000)}分钟后`;
} else if (diff < 86400000) {
workStatus.pendingQueue.nextExecuteTimeText = `${Math.floor(diff / 3600000)}小时后`;
} else {
workStatus.pendingQueue.nextExecuteTimeText = nextTime.toLocaleString('zh-CN', {
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit'
});
}
} else {
workStatus.pendingQueue.nextExecuteTimeText = '暂无';
}
// 通过MQTT发布设备工作状态
// 主题格式: device_work_status_{sn_code}
const topic = `device_work_status_${sn_code}`;
const message = JSON.stringify({
action: 'device_work_status',
data: workStatus,
timestamp: new Date().toISOString()
});
await mqttClient.publish(topic, message);
// 输出日志
console.log(`[设备工作状态] 已推送到 ${sn_code}: ${workStatus.displayText}`);
} catch (error) {
// 通知失败不影响任务执行,只记录日志
console.warn(`[设备工作状态] 推送失败:`, error.message);
}
}
/**
* 格式化任务描述
* @private
*/
_formatTaskDescription(task) {
if (task.jobTitle || task.companyName) {
const jobInfo = [];
if (task.jobTitle) {
jobInfo.push(task.jobTitle);
}
if (task.companyName) {
const companyName = task.companyName.length > 20
? task.companyName.substring(0, 20) + '...'
: task.companyName;
jobInfo.push(companyName);
}
if (jobInfo.length > 0) {
return `投递职位: ${jobInfo.join(' @ ')}`;
}
}
return task.taskName || task.taskType || '未知任务';
}
/**
* 格式化指令描述
* @private
*/
_formatCommandDescription(command) {
const params = command.command_params || command.params || {};
let parsedParams = {};
if (typeof params === 'string') {
try {
parsedParams = JSON.parse(params);
} catch (e) {
// 解析失败,忽略
}
} else {
parsedParams = params;
}
// 根据指令类型格式化描述
const commandType = command.command_type || command.type || '';
const commandName = command.command_name || command.name || '';
if (parsedParams.jobTitle && parsedParams.companyName) {
const companyName = parsedParams.companyName.length > 20
? parsedParams.companyName.substring(0, 20) + '...'
: parsedParams.companyName;
return `投递职位: ${parsedParams.jobTitle} @ ${companyName}`;
} else if (parsedParams.jobTitle) {
return `投递职位: ${parsedParams.jobTitle}`;
} else if (commandType === 'applyJob' || commandName.includes('投递')) {
return '投递简历';
} else if (commandType === 'searchJobs' || commandName.includes('搜索')) {
return `搜索职位: ${parsedParams.keyword || '默认关键词'}`;
} else if (commandType === 'sendChatMessage' || commandName.includes('沟通')) {
return '发送消息';
} else if (commandName) {
return commandName;
}
return '执行指令';
}
/**
* 格式化整体显示文案(服务端统一处理,客户端直接显示)
* @private
*/
_formatDisplayText(workStatus) {
const parts = [];
// 当前活动
if (workStatus.currentActivity) {
const activity = workStatus.currentActivity;
const typeText = activity.type === 'command' ? '指令' : '任务';
const statusText = activity.status === 'running' ? '执行中' :
activity.status === 'completed' ? '已完成' :
activity.status === 'failed' ? '失败' : '未知';
parts.push(`${typeText}: ${activity.description || activity.name} (${statusText}, 进度: ${activity.progress}%)`);
} else {
parts.push('当前活动: 无');
}
// 待执行数量
parts.push(`待执行: ${workStatus.pendingQueue.totalCount}`);
// 下次执行时间
parts.push(`下次执行: ${workStatus.pendingQueue.nextExecuteTimeText || '暂无'}`);
return parts.join(' | ');
}
}
// 导出单例
const deviceWorkStatusNotifier = new DeviceWorkStatusNotifier();
module.exports = deviceWorkStatusNotifier;

View File

@@ -11,6 +11,7 @@ const utils = require('./utils.js');
const TaskHandlers = require('./taskHandlers.js');
const MqttDispatcher = require('../mqtt/mqttDispatcher.js');
const ScheduledJobs = require('./scheduledJobs.js');
const DeviceWorkStatusNotifier = require('./deviceWorkStatusNotifier.js');
/**
* 调度系统管理器
@@ -69,6 +70,8 @@ class ScheduleManager {
*/
async initMqttClient() {
this.mqttClient = await mqttManager.getInstance();
// 设置设备工作状态推送服务的 MQTT 客户端
DeviceWorkStatusNotifier.setMqttClient(this.mqttClient);
}
/**

View File

@@ -331,11 +331,13 @@ class ScheduledJobs {
continue;
}
// 设备在线,发送任务状态摘要
// 设备在线,推送设备工作状态
try {
await this.taskQueue.sendTaskStatusSummary(sn_code);
const deviceWorkStatusNotifier = require('./deviceWorkStatusNotifier');
const summary = await this.taskQueue.getTaskStatusSummary(sn_code);
await deviceWorkStatusNotifier.sendDeviceWorkStatus(sn_code, summary);
} catch (error) {
console.error(`[任务状态同步] 设备 ${sn_code} 同步失败:`, error.message);
console.error(`[设备工作状态同步] 设备 ${sn_code} 同步失败:`, error.message);
}
}
} catch (error) {

View File

@@ -582,15 +582,10 @@ class TaskQueue {
{ where: { id: task.id } }
);
// 通知客户端任务状态变更
await this.notifyTaskStatusChange(task.sn_code, {
taskId: task.id,
taskName: task.taskName,
taskType: task.taskType,
status: 'running',
progress: 0,
startTime: task.startTime
});
// 推送设备工作状态(任务开始执行)
const deviceWorkStatusNotifier = require('./deviceWorkStatusNotifier');
const summary = await this.getTaskStatusSummary(task.sn_code);
await deviceWorkStatusNotifier.sendDeviceWorkStatus(task.sn_code, summary);
// 使用注册的任务处理器执行任务
const handler = this.taskHandlers.get(task.taskType);
@@ -627,15 +622,10 @@ class TaskQueue {
{ where: { id: task.id } }
);
// 通知客户端任务状态变更
await this.notifyTaskStatusChange(task.sn_code, {
taskId: task.id,
taskName: task.taskName,
taskType: task.taskType,
status: 'completed',
progress: 100,
endTime: task.endTime
});
// 推送设备工作状态(任务完成)
const deviceWorkStatusNotifier = require('./deviceWorkStatusNotifier');
const summary = await this.getTaskStatusSummary(task.sn_code);
await deviceWorkStatusNotifier.sendDeviceWorkStatus(task.sn_code, summary);
console.log(`[任务队列] 设备 ${task.sn_code} 任务执行成功: ${task.taskName} (耗时: ${task.duration}ms)`);
@@ -671,16 +661,10 @@ class TaskQueue {
{ where: { id: task.id } }
);
// 通知客户端任务状态变更
await this.notifyTaskStatusChange(task.sn_code, {
taskId: task.id,
taskName: task.taskName,
taskType: task.taskType,
status: 'failed',
progress: 0,
errorMessage: task.errorMessage,
endTime: task.endTime
});
// 推送设备工作状态(任务失败)
const deviceWorkStatusNotifier = require('./deviceWorkStatusNotifier');
const summary = await this.getTaskStatusSummary(task.sn_code);
await deviceWorkStatusNotifier.sendDeviceWorkStatus(task.sn_code, summary);
} catch (dbError) {
console.error(`[任务队列] 更新任务失败状态到数据库失败:`, dbError);
}
@@ -1026,34 +1010,7 @@ class TaskQueue {
}
}
/**
* 通知客户端任务状态变更
* @param {string} sn_code - 设备SN码
* @param {object} taskData - 任务数据
*/
async notifyTaskStatusChange(sn_code, taskData) {
try {
const mqttClient = await this.getMqttClient();
if (!mqttClient) {
return; // MQTT客户端不可用静默失败
}
// 通过MQTT发布任务状态变更通知
// 主题格式: task_status_{sn_code}
const topic = `task_status_${sn_code}`;
const message = JSON.stringify({
action: 'task_status_update',
data: taskData,
timestamp: new Date().toISOString()
});
await mqttClient.publish(topic, message);
console.log(`[任务队列] 已通知客户端任务状态变更: ${sn_code} - ${taskData.taskName || taskData.taskType || '未知任务'} (${taskData.status})`);
} catch (error) {
// 通知失败不影响任务执行,只记录日志
console.warn(`[任务队列] 通知客户端任务状态变更失败:`, error.message);
}
}
// notifyTaskStatusChange 方法已移除,统一使用 deviceWorkStatusNotifier.sendDeviceWorkStatus
/**
* 获取任务状态摘要(用于同步到客户端)
@@ -1260,74 +1217,12 @@ class TaskQueue {
}
/**
* 向客户端发送任务状态摘要
* 发送设备工作状态(统一方法,包含任务和指令状态)
* @param {string} sn_code - 设备SN码
* @param {object} options - 可选参数
* @param {object} options.currentCommand - 当前执行的指令信息(可选)
*/
async sendTaskStatusSummary(sn_code) {
try {
const mqttClient = await this.getMqttClient();
if (!mqttClient) {
return; // MQTT客户端不可用静默失败
}
const summary = await this.getTaskStatusSummary(sn_code);
// 通过MQTT发布任务状态摘要
// 主题格式: task_status_{sn_code}
const topic = `task_status_${sn_code}`;
const message = JSON.stringify({
action: 'task_status_summary',
data: summary,
timestamp: new Date().toISOString()
});
await mqttClient.publish(topic, message);
// 改进日志输出,显示更详细的任务信息
if (summary.currentTask) {
const task = summary.currentTask;
// 构建任务标识信息
let taskIdentifier = task.taskName || task.taskType || '未知任务';
// 如果是投递简历任务,显示具体的职位和公司信息
if (task.jobTitle || task.companyName) {
const jobInfo = [];
if (task.jobTitle) {
jobInfo.push(task.jobTitle);
}
if (task.companyName) {
// 截断过长的公司名称
const companyName = task.companyName.length > 20
? task.companyName.substring(0, 20) + '...'
: task.companyName;
jobInfo.push(companyName);
}
if (jobInfo.length > 0) {
taskIdentifier = `投递职位: ${jobInfo.join(' @ ')}`;
}
}
const progressInfo = task.progress !== null && task.progress !== undefined
? `进度: ${task.progress}%`
: '';
const stepInfo = task.currentStep
? `步骤: ${task.currentStep}`
: '';
const detailInfo = [progressInfo, stepInfo].filter(Boolean).join(', ');
const detailStr = detailInfo ? ` (${detailInfo})` : '';
// 使用总待执行数(包括当前任务的剩余步骤)
const totalCount = summary.totalPendingCount !== undefined ? summary.totalPendingCount : summary.pendingCount;
console.log(`[任务队列] 已发送任务状态摘要到 ${sn_code}: 当前任务=${taskIdentifier}${detailStr}, 待执行=${totalCount}`);
} else {
console.log(`[任务队列] 已发送任务状态摘要到 ${sn_code}: 当前任务=无, 待执行=${summary.pendingCount}`);
}
} catch (error) {
// 通知失败不影响任务执行,只记录日志
console.warn(`[任务队列] 发送任务状态摘要失败:`, error.message);
}
}
}
// 导出单例