696 lines
32 KiB
PHP
696 lines
32 KiB
PHP
<?php
|
||
|
||
namespace app\process;
|
||
|
||
use Workerman\Worker;
|
||
use Workerman\Timer;
|
||
use app\service\DataSource\DataSourceAdapterFactory;
|
||
use app\service\DataCollectionTaskService;
|
||
use app\service\DataSourceService;
|
||
use app\repository\DataCollectionTaskRepository;
|
||
use app\repository\DataSourceRepository;
|
||
use app\utils\QueueService;
|
||
use app\utils\RedisHelper;
|
||
use app\utils\LoggerHelper;
|
||
use Cron\CronExpression;
|
||
|
||
/**
|
||
* 数据采集任务调度器
|
||
*
|
||
* 职责：
|
||
* - 定时执行数据采集任务
|
||
* - 多进程并行处理（每个 Worker 处理分配给它的任务）
|
||
* - 使用分布式锁防止重复执行
|
||
* - 调用业务处理类执行数据采集逻辑
|
||
* - 支持从配置文件和数据库读取任务
|
||
*/
|
||
class DataSyncScheduler
|
||
{
|
||
private array $tasks = [];
|
||
private array $globalConfig = [];
|
||
private array $dataSourcesConfig = []; // 缓存的数据源配置（从数据库读取）
|
||
private DataCollectionTaskService $taskService;
|
||
private DataSourceService $dataSourceService;
|
||
private ?\app\service\TagTaskService $tagTaskService = null; // 延迟初始化
|
||
private array $runningTasks = []; // 正在运行的任务（实时监听模式）
|
||
|
||
public function onWorkerStart(Worker $worker): void
|
||
{
|
||
// 初始化任务服务
|
||
$this->taskService = new DataCollectionTaskService(
|
||
new DataCollectionTaskRepository()
|
||
);
|
||
|
||
// 初始化数据源服务
|
||
$this->dataSourceService = new DataSourceService(
|
||
new DataSourceRepository()
|
||
);
|
||
|
||
// 初始化标签任务服务（避免在多个方法中重复实例化）
|
||
$this->tagTaskService = new \app\service\TagTaskService(
|
||
new \app\repository\TagTaskRepository(),
|
||
new \app\repository\TagTaskExecutionRepository(),
|
||
new \app\repository\UserProfileRepository(),
|
||
new \app\service\TagService(
|
||
new \app\repository\TagDefinitionRepository(),
|
||
new \app\repository\UserProfileRepository(),
|
||
new \app\repository\UserTagRepository(),
|
||
new \app\repository\TagHistoryRepository(),
|
||
new \app\service\TagRuleEngine\SimpleRuleEngine()
|
||
)
|
||
);
|
||
|
||
// 加载任务采集配置（配置文件中的任务）
|
||
$taskConfig = config('data_collection_tasks', []);
|
||
$this->tasks = $taskConfig['tasks'] ?? [];
|
||
$this->globalConfig = $taskConfig['global'] ?? [];
|
||
|
||
// 从数据库加载数据源配置（替代config('data_sources')）
|
||
$this->loadDataSourcesConfig();
|
||
|
||
LoggerHelper::logBusiness('data_collection_scheduler_started', [
|
||
'worker_id' => $worker->id,
|
||
'total_workers' => $worker->count,
|
||
'config_task_count' => count($this->tasks),
|
||
'data_source_count' => count($this->dataSourcesConfig),
|
||
]);
|
||
|
||
// 加载配置文件中的任务
|
||
$this->loadConfigTasks($worker);
|
||
|
||
// 加载数据库中的动态任务
|
||
$this->loadDatabaseTasks($worker);
|
||
|
||
// 每30秒刷新一次数据库任务列表（检查新任务、状态变更等）
|
||
Timer::add(30, function () use ($worker) {
|
||
$this->refreshDatabaseTasks($worker);
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 从数据库加载数据源配置
|
||
*/
|
||
private function loadDataSourcesConfig(): void
|
||
{
|
||
try {
|
||
$this->dataSourcesConfig = $this->dataSourceService->getAllEnabledDataSources();
|
||
} catch (\Throwable $e) {
|
||
LoggerHelper::logError($e, [
|
||
'component' => 'DataSyncScheduler',
|
||
'action' => 'loadDataSourcesConfig',
|
||
]);
|
||
// 如果加载失败，使用空数组，避免后续错误
|
||
$this->dataSourcesConfig = [];
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 加载配置文件中的任务
|
||
*/
|
||
private function loadConfigTasks(Worker $worker): void
|
||
{
|
||
// 为每个启用的任务设置定时任务
|
||
foreach ($this->tasks as $taskId => $taskConfig) {
|
||
if (!($taskConfig['enabled'] ?? true)) {
|
||
continue;
|
||
}
|
||
|
||
// 检查是否应该由当前 Worker 处理（分片分配）
|
||
if (!$this->shouldHandleTask($taskId, $taskConfig, $worker)) {
|
||
continue;
|
||
}
|
||
|
||
// 获取调度配置
|
||
$schedule = $taskConfig['schedule'] ?? [];
|
||
|
||
// 如果调度被禁用，直接启动任务（持续运行）
|
||
if (!($schedule['enabled'] ?? true)) {
|
||
// 使用 Timer 延迟启动，避免阻塞 Worker 启动
|
||
Timer::add(0, function () use ($taskId, $taskConfig, $worker) {
|
||
$this->executeTask($taskId, $taskConfig, $worker);
|
||
}, [], false);
|
||
continue;
|
||
}
|
||
|
||
$cronExpression = $schedule['cron'] ?? '*/5 * * * *'; // 默认每5分钟
|
||
|
||
// 创建定时任务
|
||
$this->scheduleTask($taskId, $taskConfig, $cronExpression, $worker);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 加载数据库中的动态任务
|
||
*/
|
||
private function loadDatabaseTasks(Worker $worker): void
|
||
{
|
||
try {
|
||
// 加载数据采集任务
|
||
$dataCollectionTaskService = new \app\service\DataCollectionTaskService(
|
||
new \app\repository\DataCollectionTaskRepository()
|
||
);
|
||
$runningCollectionTasks = $dataCollectionTaskService->getRunningTasks();
|
||
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 从数据库加载到 " . count($runningCollectionTasks) . " 个运行中的任务 (worker_id={$worker->id})\n");
|
||
|
||
foreach ($runningCollectionTasks as $task) {
|
||
$taskId = $task['task_id'];
|
||
$taskName = $task['name'] ?? $taskId;
|
||
|
||
if (!$this->shouldHandleDatabaseTask($taskId, $worker)) {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 跳过任务 [{$taskName}]，由其他 Worker 处理 (worker_id={$worker->id})\n");
|
||
continue;
|
||
}
|
||
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] ✓ 准备启动任务: [{$taskName}] task_id={$taskId} (worker_id={$worker->id})\n");
|
||
|
||
$taskConfig = $this->convertDatabaseTaskToConfig($task);
|
||
|
||
// 使用统一的任务启动方法
|
||
$this->startTask($taskId, $taskConfig, $worker);
|
||
}
|
||
|
||
// 加载标签任务（使用已初始化的服务）
|
||
$runningTagTasks = $this->tagTaskService->getTaskList(['status' => 'running'], 1, 1000);
|
||
|
||
foreach ($runningTagTasks['tasks'] as $task) {
|
||
$taskId = $task['task_id'];
|
||
|
||
if (!$this->shouldHandleDatabaseTask($taskId, $worker)) {
|
||
continue;
|
||
}
|
||
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] ✓ 准备启动标签任务: task_id={$taskId} (worker_id={$worker->id})\n");
|
||
|
||
$taskConfig = $this->convertDatabaseTaskToConfig($task);
|
||
|
||
// 标签任务通常是批量执行，根据调度配置执行
|
||
$schedule = $taskConfig['schedule'] ?? [];
|
||
if (!($schedule['enabled'] ?? true)) {
|
||
// 立即执行一次
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] → 立即执行标签任务\n");
|
||
Timer::add(0, function () use ($taskId, $taskConfig, $worker) {
|
||
$this->executeTask($taskId, $taskConfig, $worker);
|
||
}, [], false);
|
||
} else {
|
||
$cronExpression = $schedule['cron'] ?? '0 2 * * *'; // 默认每天凌晨2点
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] → 定时执行标签任务，Cron: {$cronExpression}\n");
|
||
$this->scheduleTask($taskId, $taskConfig, $cronExpression, $worker);
|
||
}
|
||
}
|
||
} catch (\Throwable $e) {
|
||
LoggerHelper::logError($e, [
|
||
'component' => 'DataSyncScheduler',
|
||
'action' => 'loadDatabaseTasks',
|
||
]);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 刷新数据库任务列表
|
||
*/
|
||
private function refreshDatabaseTasks(Worker $worker): void
|
||
{
|
||
try {
|
||
// 刷新数据采集任务
|
||
$dataCollectionTaskService = new \app\service\DataCollectionTaskService(
|
||
new \app\repository\DataCollectionTaskRepository()
|
||
);
|
||
$runningCollectionTasks = $dataCollectionTaskService->getRunningTasks();
|
||
$collectionTaskIds = array_column($runningCollectionTasks, 'task_id');
|
||
|
||
foreach ($runningCollectionTasks as $task) {
|
||
$taskId = $task['task_id'];
|
||
|
||
if (isset($this->runningTasks[$taskId])) {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 任务 {$taskId} 已在运行中，跳过\n");
|
||
continue;
|
||
}
|
||
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 检测到新的运行中任务: {$taskId}\n");
|
||
|
||
$taskConfig = $this->convertDatabaseTaskToConfig($task);
|
||
|
||
// 使用统一的任务启动方法
|
||
$this->startTask($taskId, $taskConfig, $worker);
|
||
}
|
||
|
||
// 刷新标签任务（使用已初始化的服务）
|
||
$runningTagTasks = $this->tagTaskService->getTaskList(['status' => 'running'], 1, 1000);
|
||
$tagTaskIds = array_column($runningTagTasks['tasks'], 'task_id');
|
||
|
||
foreach ($runningTagTasks['tasks'] as $task) {
|
||
$taskId = $task['task_id'];
|
||
|
||
if (isset($this->runningTasks[$taskId])) {
|
||
continue;
|
||
}
|
||
|
||
if (RedisHelper::exists("tag_task:{$taskId}:start")) {
|
||
RedisHelper::del("tag_task:{$taskId}:start");
|
||
|
||
$taskConfig = $this->convertDatabaseTaskToConfig($task);
|
||
|
||
Timer::add(0, function () use ($taskId, $taskConfig, $worker) {
|
||
$this->executeTask($taskId, $taskConfig, $worker);
|
||
}, [], false);
|
||
$this->runningTasks[$taskId] = true;
|
||
}
|
||
}
|
||
|
||
// 检查是否有任务需要停止或暂停
|
||
$allTaskIds = array_merge($collectionTaskIds, $tagTaskIds);
|
||
foreach (array_keys($this->runningTasks) as $taskId) {
|
||
if (!in_array($taskId, $allTaskIds)) {
|
||
// 任务不在运行列表中了，移除
|
||
unset($this->runningTasks[$taskId]);
|
||
} elseif (RedisHelper::exists("data_collection_task:{$taskId}:stop") ||
|
||
RedisHelper::exists("tag_task:{$taskId}:stop")) {
|
||
// 检查停止标志
|
||
RedisHelper::del("data_collection_task:{$taskId}:stop");
|
||
RedisHelper::del("tag_task:{$taskId}:stop");
|
||
unset($this->runningTasks[$taskId]);
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 检测到停止信号，任务 {$taskId} 已停止\n");
|
||
} elseif (RedisHelper::exists("data_collection_task:{$taskId}:pause")) {
|
||
// 检查暂停标志
|
||
RedisHelper::del("data_collection_task:{$taskId}:pause");
|
||
unset($this->runningTasks[$taskId]);
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 检测到暂停信号，任务 {$taskId} 已暂停（正在执行的任务会在下次检查时停止）\n");
|
||
}
|
||
}
|
||
} catch (\Throwable $e) {
|
||
LoggerHelper::logError($e, [
|
||
'component' => 'DataSyncScheduler',
|
||
'action' => 'refreshDatabaseTasks',
|
||
]);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 判断是否应该处理数据库任务
|
||
*/
|
||
private function shouldHandleDatabaseTask(string $taskId, Worker $worker): bool
|
||
{
|
||
// 简单策略：按 Worker ID 取模分配
|
||
// 可以根据需要调整分配策略
|
||
return ($worker->id % $worker->count) === (hexdec(substr($taskId, 0, 8)) % $worker->count);
|
||
}
|
||
|
||
/**
|
||
* 将数据库任务转换为配置格式
|
||
*/
|
||
private function convertDatabaseTaskToConfig(array $task): array
|
||
{
|
||
// 判断任务类型：数据采集任务还是标签任务
|
||
// 如果任务有 data_source_id，说明是数据采集任务
|
||
// 如果任务有 target_tag_ids，说明是标签任务
|
||
if (isset($task['data_source_id']) && !empty($task['data_source_id'])) {
|
||
// 数据采集任务
|
||
// 根据 target_type 选择不同的 Handler
|
||
$targetType = $task['target_type'] ?? 'generic';
|
||
$handlerClass = \app\service\DataCollection\Handler\GenericCollectionHandler::class;
|
||
|
||
if ($targetType === 'consumption_record') {
|
||
$handlerClass = \app\service\DataCollection\Handler\ConsumptionCollectionHandler::class;
|
||
}
|
||
|
||
$config = [
|
||
'task_id' => $task['task_id'],
|
||
'name' => $task['name'] ?? '',
|
||
'data_source_id' => $task['data_source_id'],
|
||
'data_source' => $task['data_source_id'],
|
||
'database' => $task['database'] ?? '',
|
||
'collection' => $task['collection'] ?? null,
|
||
'collections' => $task['collections'] ?? null,
|
||
'target_type' => $targetType,
|
||
'target_data_source_id' => $task['target_data_source_id'] ?? null,
|
||
'target_database' => $task['target_database'] ?? null,
|
||
'target_collection' => $task['target_collection'] ?? null,
|
||
'mode' => $task['mode'] ?? 'batch',
|
||
'field_mappings' => $task['field_mappings'] ?? [],
|
||
'collection_field_mappings' => $task['collection_field_mappings'] ?? [],
|
||
'lookups' => $task['lookups'] ?? [],
|
||
'collection_lookups' => $task['collection_lookups'] ?? [],
|
||
'filter_conditions' => $task['filter_conditions'] ?? [],
|
||
'schedule' => $task['schedule'] ?? [
|
||
'enabled' => false,
|
||
'cron' => null,
|
||
],
|
||
'handler_class' => $handlerClass,
|
||
'batch_size' => $task['batch_size'] ?? 1000,
|
||
];
|
||
|
||
// 对于consumption_record类型的任务，添加source_type字段（如果存在）
|
||
if ($targetType === 'consumption_record' && isset($task['source_type'])) {
|
||
$config['source_type'] = $task['source_type'];
|
||
}
|
||
|
||
return $config;
|
||
} elseif (isset($task['target_tag_ids']) || isset($task['task_type'])) {
|
||
// 标签任务
|
||
return [
|
||
'task_id' => $task['task_id'],
|
||
'name' => $task['name'] ?? '',
|
||
'task_type' => $task['task_type'] ?? 'full',
|
||
'target_tag_ids' => $task['target_tag_ids'] ?? [],
|
||
'user_scope' => $task['user_scope'] ?? ['type' => 'all'],
|
||
'schedule' => $task['schedule'] ?? [
|
||
'enabled' => false,
|
||
'cron' => null,
|
||
],
|
||
'config' => $task['config'] ?? [
|
||
'concurrency' => 10,
|
||
'batch_size' => 100,
|
||
'error_handling' => 'skip',
|
||
],
|
||
'handler_class' => \app\service\DataCollection\Handler\TagTaskHandler::class,
|
||
];
|
||
} else {
|
||
throw new \InvalidArgumentException("无法识别任务类型: {$task['task_id']}");
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 判断当前 Worker 是否应该处理该任务（分片分配）
|
||
*
|
||
* @param string $taskId 任务ID
|
||
* @param array<string, mixed> $taskConfig 任务配置
|
||
* @param Worker $worker Worker 实例
|
||
* @return bool 是否应该处理
|
||
*/
|
||
private function shouldHandleTask(string $taskId, array $taskConfig, Worker $worker): bool
|
||
{
|
||
$sharding = $taskConfig['sharding'] ?? [];
|
||
$strategy = $sharding['strategy'] ?? 'none';
|
||
|
||
// 如果不需要分片，所有 Worker 都处理（但通过分布式锁保证只有一个执行）
|
||
if ($strategy === 'none') {
|
||
return true;
|
||
}
|
||
|
||
// by_database 策略：所有 Worker 都处理，但每个 Worker 处理不同的数据库（在 Handler 中分配）
|
||
if ($strategy === 'by_database') {
|
||
return true; // 所有 Worker 都处理，数据库分配在 Handler 中进行
|
||
}
|
||
|
||
// 其他分片策略：按 Worker ID 取模分配
|
||
$shardCount = $sharding['shard_count'] ?? 1;
|
||
if ($shardCount <= 1) {
|
||
// 如果 shard_count <= 1，所有 Worker 都处理
|
||
return true;
|
||
}
|
||
|
||
$assignedShardId = $worker->id % $shardCount;
|
||
// 对于分片策略，只处理分配给当前 Worker 的分片
|
||
return $assignedShardId === ($worker->id % $shardCount);
|
||
}
|
||
|
||
/**
|
||
* 为任务设置定时任务
|
||
*
|
||
* @param string $taskId 任务ID
|
||
* @param array<string, mixed> $taskConfig 任务配置
|
||
* @param string $cronExpression Cron 表达式
|
||
* @param Worker $worker Worker 实例
|
||
* @return void
|
||
*/
|
||
private function scheduleTask(string $taskId, array $taskConfig, string $cronExpression, Worker $worker): void
|
||
{
|
||
try {
|
||
$cron = CronExpression::factory($cronExpression);
|
||
|
||
// 计算下次执行时间
|
||
$nextRunTime = $cron->getNextRunDate()->getTimestamp();
|
||
$now = time();
|
||
$delay = max(0, $nextRunTime - $now);
|
||
|
||
LoggerHelper::logBusiness('data_collection_task_scheduled', [
|
||
'task_id' => $taskId,
|
||
'task_name' => $taskConfig['name'] ?? $taskId,
|
||
'cron' => $cronExpression,
|
||
'next_run_time' => date('Y-m-d H:i:s', $nextRunTime),
|
||
'delay' => $delay,
|
||
]);
|
||
|
||
// 设置定时器（延迟执行第一次，然后每60秒检查一次）
|
||
Timer::add(60, function () use ($taskId, $taskConfig, $cron, $worker) {
|
||
$now = time();
|
||
$nextRunTime = $cron->getNextRunDate()->getTimestamp();
|
||
|
||
// 如果到了执行时间（允许1秒误差）
|
||
if ($nextRunTime <= $now + 1) {
|
||
$this->executeTask($taskId, $taskConfig, $worker);
|
||
}
|
||
}, [], false, $delay);
|
||
} catch (\Throwable $e) {
|
||
LoggerHelper::logError($e, [
|
||
'component' => 'DataSyncScheduler',
|
||
'action' => 'scheduleTask',
|
||
'task_id' => $taskId,
|
||
'cron' => $cronExpression,
|
||
]);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 执行采集任务
|
||
*
|
||
* @param string $taskId 任务ID
|
||
* @param array<string, mixed> $taskConfig 任务配置
|
||
* @param Worker $worker Worker 实例
|
||
* @return void
|
||
*/
|
||
private function executeTask(string $taskId, array $taskConfig, Worker $worker): void
|
||
{
|
||
// 在执行前，检查任务状态，如果已经是 completed，就不应该再执行
|
||
try {
|
||
$task = $this->taskService->getTask($taskId);
|
||
if ($task && isset($task['status'])) {
|
||
if ($task['status'] === 'completed') {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] ⚠️ 任务已完成，跳过执行: task_id={$taskId}\n");
|
||
LoggerHelper::logBusiness('data_collection_skipped_completed', [
|
||
'task_id' => $taskId,
|
||
'worker_id' => $worker->id,
|
||
]);
|
||
return;
|
||
}
|
||
if (in_array($task['status'], ['stopped', 'paused', 'error'])) {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] ⚠️ 任务状态为 {$task['status']}，跳过执行: task_id={$taskId}\n");
|
||
LoggerHelper::logBusiness('data_collection_skipped_status', [
|
||
'task_id' => $taskId,
|
||
'status' => $task['status'],
|
||
'worker_id' => $worker->id,
|
||
]);
|
||
return;
|
||
}
|
||
}
|
||
} catch (\Throwable $e) {
|
||
// 如果查询任务状态失败，记录日志但继续执行（避免因为查询失败而阻塞任务）
|
||
LoggerHelper::logError($e, [
|
||
'component' => 'DataSyncScheduler',
|
||
'action' => 'executeTask',
|
||
'task_id' => $taskId,
|
||
'message' => '检查任务状态失败，继续执行任务',
|
||
]);
|
||
}
|
||
|
||
$sharding = $taskConfig['sharding'] ?? [];
|
||
$strategy = $sharding['strategy'] ?? 'none';
|
||
|
||
// 对于 by_database 策略，不使用全局锁，让所有 Worker 并行执行
|
||
// 每个 Worker 会在 Handler 中分配不同的数据库
|
||
$useLock = ($strategy !== 'by_database');
|
||
|
||
if ($useLock) {
|
||
$lockKey = "data_collection:{$taskId}";
|
||
$lockConfig = $this->globalConfig['distributed_lock'] ?? [];
|
||
$ttl = $lockConfig['ttl'] ?? 300;
|
||
$retryTimes = $lockConfig['retry_times'] ?? 3;
|
||
$retryDelay = $lockConfig['retry_delay'] ?? 1000;
|
||
|
||
// 尝试获取分布式锁
|
||
if (!RedisHelper::acquireLock($lockKey, $ttl, $retryTimes, $retryDelay)) {
|
||
LoggerHelper::logBusiness('data_collection_skipped_locked', [
|
||
'task_id' => $taskId,
|
||
'worker_id' => $worker->id,
|
||
]);
|
||
return;
|
||
}
|
||
}
|
||
|
||
try {
|
||
LoggerHelper::logBusiness('data_collection_started', [
|
||
'task_id' => $taskId,
|
||
'task_name' => $taskConfig['name'] ?? $taskId,
|
||
'worker_id' => $worker->id,
|
||
]);
|
||
// 控制台提示，标记具体哪个采集/同步任务被当前 Worker 拉起，方便排查任务是否真正执行
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤1-任务启动】执行采集任务：task_id={$taskId}, task_name=" . ($taskConfig['name'] ?? $taskId) . ", worker_id={$worker->id}\n");
|
||
|
||
// 判断任务类型：数据采集任务需要数据源适配器，标签任务不需要
|
||
$handlerClass = $taskConfig['handler_class'] ?? '';
|
||
$isTagTask = strpos($handlerClass, 'TagTaskHandler') !== false;
|
||
|
||
$adapter = null;
|
||
if (!$isTagTask) {
|
||
// 数据采集任务：需要数据源适配器
|
||
// 支持两种配置方式：
|
||
// 1. 单数据源：data_source（用于普通采集任务）
|
||
// 2. 多数据源：source_data_source 和 target_data_source（用于数据库同步任务）
|
||
// 3. 动态任务：data_source_id（从数据库读取的任务）
|
||
$dataSourceId = $taskConfig['data_source'] ?? $taskConfig['data_source_id'] ?? '';
|
||
$sourceDataSourceId = $taskConfig['source_data_source'] ?? '';
|
||
|
||
// 如果配置了 source_data_source，说明是多数据源任务（如数据库同步）
|
||
// 这种情况下，只需要创建源数据源适配器，目标数据源由 Handler 内部处理
|
||
if (!empty($sourceDataSourceId)) {
|
||
$dataSourceId = $sourceDataSourceId;
|
||
}
|
||
|
||
// 从缓存或数据库获取数据源配置
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤2-查询数据源配置】开始查询数据源配置: data_source_id={$dataSourceId}\n");
|
||
$dataSourceConfig = $this->dataSourcesConfig[$dataSourceId] ?? null;
|
||
|
||
// 如果缓存中没有，尝试从数据库加载
|
||
if (empty($dataSourceConfig)) {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤2-查询数据源配置】缓存中没有，从数据库加载: data_source_id={$dataSourceId}\n");
|
||
$dataSourceConfig = $this->dataSourceService->getDataSourceConfigById($dataSourceId);
|
||
if ($dataSourceConfig) {
|
||
// 更新缓存（使用原始 dataSourceId 作为 key）
|
||
$this->dataSourcesConfig[$dataSourceId] = $dataSourceConfig;
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤2-查询数据源配置】✓ 数据源配置加载成功: host={$dataSourceConfig['host']}, port={$dataSourceConfig['port']}, database={$dataSourceConfig['database']}\n");
|
||
} else {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤2-查询数据源配置】✗ 数据源配置不存在: data_source_id={$dataSourceId}\n");
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤2-查询数据源配置】提示：请检查 data_sources 表中是否存在该数据源，或检查 name 字段是否匹配\n");
|
||
}
|
||
} else {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤2-查询数据源配置】✓ 从缓存获取数据源配置: host={$dataSourceConfig['host']}, port={$dataSourceConfig['port']}\n");
|
||
}
|
||
|
||
if (empty($dataSourceConfig)) {
|
||
throw new \InvalidArgumentException("数据源配置不存在: {$dataSourceId}");
|
||
}
|
||
|
||
// 创建数据源适配器
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤3-创建数据源适配器】开始创建适配器: type={$dataSourceConfig['type']}\n");
|
||
$adapter = DataSourceAdapterFactory::create(
|
||
$dataSourceConfig['type'],
|
||
$dataSourceConfig
|
||
);
|
||
|
||
// 确保适配器已连接
|
||
if (!$adapter->isConnected()) {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤3-创建数据源适配器】适配器未连接，尝试连接...\n");
|
||
if (!$adapter->connect($dataSourceConfig)) {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤3-创建数据源适配器】✗ 连接失败: data_source_id={$dataSourceId}\n");
|
||
throw new \RuntimeException("无法连接到数据源: {$dataSourceId}");
|
||
}
|
||
}
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤3-创建数据源适配器】✓ 数据源适配器创建并连接成功\n");
|
||
}
|
||
|
||
// 创建业务处理类（处理逻辑在业务代码中）
|
||
$handlerClass = $taskConfig['handler_class'] ?? '';
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤4-创建Handler】开始创建Handler: handler_class={$handlerClass}\n");
|
||
if (empty($handlerClass) || !class_exists($handlerClass)) {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤4-创建Handler】✗ Handler类不存在: {$handlerClass}\n");
|
||
throw new \InvalidArgumentException("处理类不存在或未配置: {$handlerClass}");
|
||
}
|
||
|
||
// 实例化处理类
|
||
$handler = new $handlerClass();
|
||
|
||
// 检查处理类是否实现了采集接口
|
||
if (!method_exists($handler, 'collect')) {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤4-创建Handler】✗ Handler未实现collect方法: {$handlerClass}\n");
|
||
throw new \InvalidArgumentException("处理类必须实现 collect 方法: {$handlerClass}");
|
||
}
|
||
|
||
// 添加任务ID到配置中
|
||
$taskConfig['task_id'] = $taskId;
|
||
|
||
// 添加 Worker 信息到配置中（用于多进程数据库分配）
|
||
$taskConfig['worker_id'] = $worker->id;
|
||
$taskConfig['worker_count'] = $worker->count;
|
||
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤4-创建Handler】✓ Handler创建成功，开始调用collect方法\n");
|
||
// 调用业务处理类的采集方法
|
||
try {
|
||
$handler->collect($adapter, $taskConfig);
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤4-创建Handler】✓ Handler.collect()方法执行完成\n");
|
||
} catch (\Throwable $collectException) {
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤4-创建Handler】✗ Handler.collect()方法执行失败: " . $collectException->getMessage() . "\n");
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【步骤4-创建Handler】异常堆栈: " . $collectException->getTraceAsString() . "\n");
|
||
throw $collectException; // 重新抛出异常，让外层catch处理
|
||
}
|
||
|
||
LoggerHelper::logBusiness('data_collection_completed', [
|
||
'task_id' => $taskId,
|
||
'task_name' => $taskConfig['name'] ?? $taskId,
|
||
]);
|
||
} catch (\Throwable $e) {
|
||
// 在控制台输出异常信息，方便排查
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【异常捕获】任务执行失败: task_id={$taskId}, error=" . $e->getMessage() . "\n");
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【异常捕获】异常文件: " . $e->getFile() . ":" . $e->getLine() . "\n");
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] 【异常捕获】异常堆栈: " . $e->getTraceAsString() . "\n");
|
||
|
||
LoggerHelper::logError($e, [
|
||
'component' => 'DataSyncScheduler',
|
||
'action' => 'executeTask',
|
||
'task_id' => $taskId,
|
||
'worker_id' => $worker->id,
|
||
]);
|
||
} finally {
|
||
// 释放锁（仅在使用锁的情况下）
|
||
if (isset($useLock) && $useLock && isset($lockKey)) {
|
||
RedisHelper::releaseLock($lockKey);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 统一的任务启动方法
|
||
*
|
||
* @param string $taskId 任务ID
|
||
* @param array<string, mixed> $taskConfig 任务配置
|
||
* @param Worker $worker Worker 实例
|
||
* @return void
|
||
*/
|
||
private function startTask(string $taskId, array $taskConfig, Worker $worker): void
|
||
{
|
||
$taskName = $taskConfig['name'] ?? $taskId;
|
||
|
||
if ($taskConfig['mode'] === 'realtime') {
|
||
// 实时模式：立即启动，持续运行
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] → 实时模式任务: [{$taskName}]\n");
|
||
Timer::add(0, function () use ($taskId, $taskConfig, $worker) {
|
||
$this->executeTask($taskId, $taskConfig, $worker);
|
||
}, [], false);
|
||
$this->runningTasks[$taskId] = true;
|
||
} else {
|
||
// 批量模式：根据调度配置执行
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] → 批量模式任务: [{$taskName}]\n");
|
||
$schedule = $taskConfig['schedule'] ?? [];
|
||
if (!($schedule['enabled'] ?? true)) {
|
||
// 调度被禁用，立即执行一次
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] → 立即执行（调度已禁用）\n");
|
||
Timer::add(0, function () use ($taskId, $taskConfig, $worker) {
|
||
$this->executeTask($taskId, $taskConfig, $worker);
|
||
}, [], false);
|
||
} else {
|
||
// 使用 Cron 表达式定时执行
|
||
$cronExpression = $schedule['cron'] ?? '*/5 * * * *';
|
||
\Workerman\Worker::safeEcho("[DataSyncScheduler] → 定时执行，Cron: {$cronExpression}\n");
|
||
$this->scheduleTask($taskId, $taskConfig, $cronExpression, $worker);
|
||
}
|
||
}
|
||
}
|
||
|
||
public function onWorkerStop(Worker $worker): void
|
||
{
|
||
LoggerHelper::logBusiness('data_collection_scheduler_stopped', [
|
||
'worker_id' => $worker->id,
|
||
]);
|
||
}
|
||
}
|