TaskExecutionBackgroundService.cs
7.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Rcs.Application.Services.Protocol;
using Rcs.Application.Shared;
using Rcs.Domain.Entities;
using Rcs.Domain.Repositories;
using TaskStatus = Rcs.Domain.Entities.TaskStatus;
namespace Rcs.Infrastructure.Services
{
/// <summary>
/// 后台任务执行服务 - 将已分配的任务下发给空闲机器人执行
/// @author zzy
/// </summary>
public class TaskExecutionBackgroundService : BackgroundService, ITaskExecutionService
{
private readonly ILogger<TaskExecutionBackgroundService> _logger;
private readonly IServiceProvider _serviceProvider;
private readonly TimeSpan _executionInterval = TimeSpan.FromSeconds(3);
private const int MaxAssignedTasksPerCycle = 10;
public TaskExecutionBackgroundService(
ILogger<TaskExecutionBackgroundService> logger,
IServiceProvider serviceProvider)
{
_logger = logger;
_serviceProvider = serviceProvider;
}
/// <summary>
/// 后台服务执行入口
/// @author zzy
/// </summary>
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
_logger.LogInformation("[任务执行] 后台任务执行服务已启动");
while (!stoppingToken.IsCancellationRequested)
{
try
{
await ExecuteTasksAsync(stoppingToken);
}
catch (Exception ex)
{
_logger.LogError(ex, "[任务执行] 执行过程发生异常");
}
await Task.Delay(_executionInterval, stoppingToken);
}
_logger.LogInformation("[任务执行] 后台任务执行服务已停止");
}
/// <summary>
/// 执行一次任务执行检查
/// @author zzy
/// </summary>
public async Task<TaskExecutionResult> ExecuteTasksAsync(CancellationToken cancellationToken = default)
{
using var scope = _serviceProvider.CreateScope();
var taskRepo = scope.ServiceProvider.GetRequiredService<IRobotTaskRepository>();
var robotRepo = scope.ServiceProvider.GetRequiredService<IRobotRepository>();
// 1. 获取已分配的任务
var assignedTasks = (await taskRepo.GetByStatusAsync(TaskStatus.Assigned, cancellationToken))
.OrderBy(t => t.Priority)
.ThenBy(t => t.CreatedAt)
.Take(MaxAssignedTasksPerCycle)
.ToList();
if (!assignedTasks.Any())
{
return new TaskExecutionResult { Success = true, ExecutedCount = 0, Message = "无待执行任务" };
}
int executedCount = 0;
foreach (var task in assignedTasks)
{
if (!task.RobotId.HasValue) continue;
// 2. 获取机器人信息
var robot = await robotRepo.GetByIdFullDataAsync(task.RobotId.Value, cancellationToken);
if (robot == null)
{
_logger.LogWarning("[任务执行] 任务 {TaskCode} 对应的机器人不存在", task.TaskCode);
continue;
}
// 3. 检查机器人是否空闲且无执行中任务
if (!IsRobotAvailable(robot))
{
_logger.LogDebug("[任务执行] 机器人 {RobotCode} 当前不可用", robot.RobotCode);
continue;
}
// 4. 检查机器人是否有执行中的任务
var hasInProgressTask = await HasInProgressTaskAsync(robot.RobotId, taskRepo, cancellationToken);
if (hasInProgressTask)
{
_logger.LogDebug("[任务执行] 机器人 {RobotCode} 存在执行中任务", robot.RobotCode);
continue;
}
// 5. 根据协议类型执行任务
var success = await ExecuteTaskByProtocolAsync(task, robot, taskRepo, scope, cancellationToken);
if (success)
{
executedCount++;
}
}
return new TaskExecutionResult
{
Success = true,
ExecutedCount = executedCount,
Message = $"本次执行完成，已下发 {executedCount} 个任务"
};
}
/// <summary>
/// 检查机器人是否可用（空闲、在线、未暂停）
/// @author zzy
/// </summary>
private bool IsRobotAvailable(Robot robot)
{
return robot.Status == RobotStatus.Idle
&& robot.Online == OnlineStatus.Online
&& robot.Active
&& !robot.Driving
&& !robot.Paused;
}
/// <summary>
/// 检查机器人是否有执行中的任务
/// @author zzy
/// </summary>
private async Task<bool> HasInProgressTaskAsync(
Guid robotId,
IRobotTaskRepository taskRepo,
CancellationToken cancellationToken)
{
var robotTasks = await taskRepo.GetByRobotIdAsync(robotId, cancellationToken);
return robotTasks.Any(t => t.IsInProgress());
}
/// <summary>
/// 根据协议类型执行任务
/// @author zzy
/// 重构为使用协议工厂模式
/// </summary>
private async Task<bool> ExecuteTaskByProtocolAsync(
RobotTask task,
Robot robot,
IRobotTaskRepository taskRepo,
IServiceScope scope,
CancellationToken cancellationToken)
{
try
{
// 使用同一scope的taskRepo获取带详情的任务，避免跨scope实体跟踪冲突
var taskWithDetails = await taskRepo.GetByIdWithDetailsAsync(task.TaskId, cancellationToken);
if (taskWithDetails == null)
{
_logger.LogWarning("[任务执行] 任务 {TaskCode} 详情获取失败", task.TaskCode);
return false;
}
// 获取协议服务工厂
var protocolServiceFactory = scope.ServiceProvider.GetRequiredService<IProtocolServiceFactory>();
var protocolService = protocolServiceFactory.GetService(robot);
// 执行任务
var res = await protocolService.SendOrderAsync(robot, taskWithDetails, cancellationToken);
if (res.Success)
{
taskWithDetails.StartExecution();
await taskRepo.UpdateAsync(taskWithDetails, cancellationToken);
await taskRepo.SaveChangesAsync(cancellationToken);
_logger.LogInformation("[任务执行] 任务 {TaskCode} 已下发给机器人 {RobotCode}，协议类型: {ProtocolType}",
taskWithDetails.TaskCode, robot.RobotCode, robot.ProtocolType);
}
return res.Success;
}
catch (Exception ex)
{
_logger.LogError(ex, "[任务执行] 任务 {TaskCode} 执行失败", task.TaskCode);
return false;
}
}
}
}