1418 lines
58 KiB
PHP
1418 lines
58 KiB
PHP
<?php
|
||
|
||
namespace app\service;
|
||
|
||
use app\utils\LoggerHelper;
|
||
use MongoDB\Client;
|
||
use MongoDB\Collection;
|
||
use MongoDB\Database;
|
||
use MongoDB\Driver\Exception\Exception as MongoDBException;
|
||
use MongoDB\Model\ChangeStreamIterator;
|
||
|
||
/**
|
||
* 数据库同步服务
|
||
*
|
||
* 职责：
|
||
* - 处理数据库之间的实时同步
|
||
* - 使用 MongoDB Change Streams 监听源数据库变化
|
||
* - 将变化同步到目标数据库
|
||
*/
|
||
class DatabaseSyncService
|
||
{
|
||
private ?Client $sourceClient = null;
|
||
private ?Client $targetClient = null;
|
||
private array $config;
|
||
private array $stats = [
|
||
'databases' => 0,
|
||
'collections' => 0,
|
||
'documents_inserted' => 0,
|
||
'documents_updated' => 0,
|
||
'documents_deleted' => 0,
|
||
'errors' => 0,
|
||
'last_sync_time' => null,
|
||
];
|
||
// 同步进度信息
|
||
private array $progress = [
|
||
'status' => 'idle', // idle, full_sync, incremental_sync, error
|
||
'current_database' => null,
|
||
'current_collection' => null,
|
||
'databases_total' => 0,
|
||
'databases_completed' => 0,
|
||
'collections_total' => 0,
|
||
'collections_completed' => 0,
|
||
// 文档级进度（行数）
|
||
'documents_total' => 0,
|
||
'documents_processed' => 0,
|
||
// 数据量级进度（基于 collStats / dbStats 估算的字节数）
|
||
'bytes_total' => 0,
|
||
// 已经清空过的目标数据库列表，避免重复清空影响断点续传
|
||
'cleared_databases' => [],
|
||
// 源端数据库的集合快照（用于检测“同名库但结构已变更/被重建”的情况）
|
||
// 结构示例：'collections_snapshot' => ['KR' => ['coll1', 'coll2', ...]]
|
||
'collections_snapshot' => [],
|
||
// 在历史进度中出现过，但当前源库已不存在的数据库（用于给出提醒）
|
||
'orphan_databases' => [],
|
||
// 断点续传检查点：按数据库/集合记录最后一个处理的 _id 和已处理数量
|
||
// 结构示例：
|
||
// 'checkpoints' => [
|
||
// 'KR_腾讯' => [
|
||
// '某集合名' => [
|
||
// 'last_id' => 'xxx',
|
||
// 'processed' => 123,
|
||
// 'completed' => false,
|
||
// ],
|
||
// ],
|
||
// ],
|
||
'checkpoints' => [],
|
||
// bytes_processed 不单独持久化，在 getProgress 中按 documents 比例动态估算
|
||
'start_time' => null,
|
||
'current_database_start_time' => null,
|
||
'estimated_time_remaining' => null,
|
||
'last_error' => null, // 记录最后一次错误信息
|
||
'error_database' => null, // 出错的数据库名称
|
||
];
|
||
|
||
/**
|
||
* 构造函数
|
||
*
|
||
* @param array<string, mixed>|null $config 配置数组，必须包含 'source' 和 'target' 数据库配置
|
||
* 如果为 null 或配置无效，将跳过数据库连接初始化（仅用于读取进度文件）
|
||
* 注意：config('database_sync') 已废弃，必须通过 DatabaseSyncHandler 传递配置
|
||
*
|
||
* @throws \InvalidArgumentException 如果配置为 null 且无效
|
||
*/
|
||
public function __construct(?array $config = null)
|
||
{
|
||
if ($config === null) {
|
||
throw new \InvalidArgumentException(
|
||
'DatabaseSyncService 必须传递配置参数。' .
|
||
'config(\'database_sync\') 已废弃，请使用 DatabaseSyncHandler 传递配置。'
|
||
);
|
||
}
|
||
$this->config = $config;
|
||
|
||
try {
|
||
// 只有在配置有效时才初始化数据库连接（用于查询进度时可能不需要连接）
|
||
if ($this->hasValidConfig()) {
|
||
$this->initClients();
|
||
LoggerHelper::logBusiness('database_sync_service_initialized', [
|
||
'source' => $this->config['source']['host'] . ':' . $this->config['source']['port'],
|
||
'target' => $this->config['target']['host'] . ':' . $this->config['target']['port'],
|
||
]);
|
||
}
|
||
$this->loadProgress();
|
||
} catch (\Exception $e) {
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_service_init_error',
|
||
]);
|
||
throw $e;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 检查配置是否有效（用于判断是否需要初始化数据库连接）
|
||
*
|
||
* @return bool
|
||
*/
|
||
private function hasValidConfig(): bool
|
||
{
|
||
$sourceHost = $this->config['source']['host'] ?? '';
|
||
$sourcePort = $this->config['source']['port'] ?? 0;
|
||
$targetHost = $this->config['target']['host'] ?? '';
|
||
$targetPort = $this->config['target']['port'] ?? 0;
|
||
|
||
return !empty($sourceHost) && $sourcePort > 0 && !empty($targetHost) && $targetPort > 0;
|
||
}
|
||
|
||
/**
|
||
* 初始化数据库连接
|
||
*/
|
||
private function initClients(): void
|
||
{
|
||
// 源数据库连接
|
||
$sourceConfig = $this->config['source'];
|
||
$sourceDsn = $this->buildDsn($sourceConfig);
|
||
$this->sourceClient = new Client($sourceDsn, $sourceConfig['options']);
|
||
|
||
// 目标数据库连接
|
||
$targetConfig = $this->config['target'];
|
||
$targetDsn = $this->buildDsn($targetConfig);
|
||
$this->targetClient = new Client($targetDsn, $targetConfig['options']);
|
||
|
||
LoggerHelper::logBusiness('database_sync_clients_initialized', [
|
||
'source' => $sourceConfig['host'] . ':' . $sourceConfig['port'],
|
||
'target' => $targetConfig['host'] . ':' . $targetConfig['port'],
|
||
]);
|
||
}
|
||
|
||
/**
|
||
* 构建 MongoDB DSN
|
||
*/
|
||
private function buildDsn(array $config): string
|
||
{
|
||
// 验证必需的配置项
|
||
$host = $config['host'] ?? '';
|
||
$port = $config['port'] ?? 0;
|
||
|
||
if (empty($host)) {
|
||
throw new \InvalidArgumentException(
|
||
'MongoDB host 配置为空。请设置环境变量 DB_SYNC_SOURCE_HOST 和 DB_SYNC_TARGET_HOST'
|
||
);
|
||
}
|
||
|
||
if (empty($port) || $port <= 0) {
|
||
throw new \InvalidArgumentException(
|
||
'MongoDB port 配置无效。请设置环境变量 DB_SYNC_SOURCE_PORT 和 DB_SYNC_TARGET_PORT'
|
||
);
|
||
}
|
||
|
||
$dsn = 'mongodb://';
|
||
if (!empty($config['username']) && !empty($config['password'])) {
|
||
$dsn .= urlencode($config['username']) . ':' . urlencode($config['password']) . '@';
|
||
}
|
||
$dsn .= $host . ':' . $port;
|
||
if (!empty($config['auth_source'])) {
|
||
$dsn .= '/?authSource=' . urlencode($config['auth_source']);
|
||
}
|
||
return $dsn;
|
||
}
|
||
|
||
/**
|
||
* 获取要同步的数据库列表
|
||
*/
|
||
public function getDatabasesToSync(): array
|
||
{
|
||
try {
|
||
$databases = $this->sourceClient->listDatabases();
|
||
$databasesToSync = [];
|
||
// 记录每个数据库的大致大小，用于排序（小库优先同步）
|
||
$databaseSizes = [];
|
||
$excludeDatabases = $this->config['sync']['exclude_databases'] ?? [];
|
||
|
||
$currentDbNames = [];
|
||
foreach ($databases as $databaseInfo) {
|
||
$dbName = (string)$databaseInfo->getName();
|
||
$currentDbNames[] = $dbName;
|
||
|
||
// 记录源端数据库的大致大小（单位：字节），用于后续排序
|
||
try {
|
||
$sizeOnDisk = method_exists($databaseInfo, 'getSizeOnDisk')
|
||
? (int)$databaseInfo->getSizeOnDisk()
|
||
: 0;
|
||
$databaseSizes[$dbName] = $sizeOnDisk;
|
||
} catch (\Throwable $e) {
|
||
$databaseSizes[$dbName] = 0;
|
||
}
|
||
|
||
// 排除系统数据库
|
||
if (in_array($dbName, $excludeDatabases)) {
|
||
continue;
|
||
}
|
||
|
||
// 如果指定了要同步的数据库列表，只同步列表中的
|
||
$syncDatabases = $this->config['sync']['databases'] ?? [];
|
||
if (!empty($syncDatabases) && !in_array($dbName, $syncDatabases)) {
|
||
continue;
|
||
}
|
||
|
||
$databasesToSync[] = $dbName;
|
||
}
|
||
|
||
// 检测历史进度中曾经同步过，但当前源库已不存在的“孤儿数据库”
|
||
$knownDbNames = array_keys($this->progress['collections_snapshot'] ?? []);
|
||
$orphanDatabases = $this->progress['orphan_databases'] ?? [];
|
||
foreach ($knownDbNames as $knownDb) {
|
||
if (!in_array($knownDb, $currentDbNames, true) && !in_array($knownDb, $orphanDatabases, true)) {
|
||
$orphanDatabases[] = $knownDb;
|
||
LoggerHelper::logBusiness('database_sync_source_database_missing', [
|
||
'database' => $knownDb,
|
||
], 'warning');
|
||
}
|
||
}
|
||
$this->progress['orphan_databases'] = $orphanDatabases;
|
||
|
||
// 更新进度信息
|
||
// 根据数据库大小排序：小的优先同步，便于尽快完成更多库，提高“完成感”
|
||
usort($databasesToSync, function (string $a, string $b) use ($databaseSizes): int {
|
||
$sizeA = $databaseSizes[$a] ?? PHP_INT_MAX;
|
||
$sizeB = $databaseSizes[$b] ?? PHP_INT_MAX;
|
||
if ($sizeA === $sizeB) {
|
||
return strcmp($a, $b);
|
||
}
|
||
return $sizeA <=> $sizeB;
|
||
});
|
||
|
||
$this->progress['databases_total'] = count($databasesToSync);
|
||
// 如果是首次获取数据库列表（start_time 为空），才重置 completed 计数，
|
||
// 避免在进程中途多次调用时把已完成的统计清零。
|
||
if ($this->progress['start_time'] === null) {
|
||
$this->progress['databases_completed'] = 0;
|
||
}
|
||
if ($this->progress['start_time'] === null) {
|
||
$this->progress['start_time'] = microtime(true);
|
||
}
|
||
$this->saveProgress();
|
||
|
||
return $databasesToSync;
|
||
} catch (MongoDBException $e) {
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_list_databases_error',
|
||
]);
|
||
return [];
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 确保目标数据库存在，如果不存在则创建
|
||
*/
|
||
private function ensureTargetDatabaseExists(string $databaseName): void
|
||
{
|
||
try {
|
||
// 检查目标数据库是否存在
|
||
$targetDatabases = $this->targetClient->listDatabases();
|
||
$databaseExists = false;
|
||
|
||
foreach ($targetDatabases as $dbInfo) {
|
||
if ($dbInfo->getName() === $databaseName) {
|
||
$databaseExists = true;
|
||
break;
|
||
}
|
||
}
|
||
|
||
// 如果数据库不存在，创建一个临时集合并插入一条记录来触发数据库创建
|
||
if (!$databaseExists) {
|
||
$targetDb = $this->targetClient->selectDatabase($databaseName);
|
||
$tempCollection = $targetDb->selectCollection('__temp_sync_init__');
|
||
|
||
// 插入一条临时记录来创建数据库
|
||
$tempCollection->insertOne(['_created' => new \MongoDB\BSON\UTCDateTime()]);
|
||
|
||
// 删除临时集合
|
||
$tempCollection->drop();
|
||
|
||
LoggerHelper::logBusiness('database_sync_database_created', [
|
||
'database' => $databaseName,
|
||
'target' => $this->config['target']['host'] . ':' . $this->config['target']['port'],
|
||
]);
|
||
}
|
||
} catch (MongoDBException $e) {
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_ensure_database_error',
|
||
'database' => $databaseName,
|
||
]);
|
||
// 不抛出异常，继续执行同步（MongoDB 会在第一次插入时自动创建数据库）
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 清空目标数据库（用于全量同步前的初始化）
|
||
*
|
||
* 注意：
|
||
* - 仅在首次同步该数据库时调用（通过 progress.cleared_databases 控制）
|
||
* - 后续断点续传时不会再次清空，避免丢失已同步的数据
|
||
*/
|
||
private function clearTargetDatabase(string $databaseName): void
|
||
{
|
||
try {
|
||
$targetDb = $this->targetClient->selectDatabase($databaseName);
|
||
$targetDb->drop();
|
||
|
||
LoggerHelper::logBusiness('database_sync_target_database_cleared', [
|
||
'database' => $databaseName,
|
||
'target' => $this->config['target']['host'] . ':' . $this->config['target']['port'],
|
||
]);
|
||
} catch (MongoDBException $e) {
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_clear_target_error',
|
||
'database' => $databaseName,
|
||
]);
|
||
// 清空失败属于严重问题，这里抛出异常，避免在脏数据基础上继续同步
|
||
throw $e;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 全量同步数据库
|
||
*/
|
||
public function fullSyncDatabase(string $databaseName): bool
|
||
{
|
||
try {
|
||
// 更新进度状态
|
||
if ($this->progress['start_time'] === null) {
|
||
$this->progress['start_time'] = microtime(true);
|
||
}
|
||
$this->progress['status'] = 'full_sync';
|
||
$this->progress['current_database'] = $databaseName;
|
||
$this->progress['current_database_start_time'] = microtime(true);
|
||
$this->saveProgress();
|
||
|
||
LoggerHelper::logBusiness('database_sync_database_start', [
|
||
'database' => $databaseName,
|
||
'status' => 'full_sync',
|
||
]);
|
||
|
||
// 确保目标数据库存在
|
||
$this->ensureTargetDatabaseExists($databaseName);
|
||
|
||
// 如果尚未清空过该目标数据库，则执行一次清空（适用于你当前“目标库可以清空”的场景）
|
||
$clearedDatabases = $this->progress['cleared_databases'] ?? [];
|
||
if (!in_array($databaseName, $clearedDatabases, true)) {
|
||
$this->clearTargetDatabase($databaseName);
|
||
$clearedDatabases[] = $databaseName;
|
||
$this->progress['cleared_databases'] = $clearedDatabases;
|
||
$this->saveProgress();
|
||
}
|
||
|
||
$sourceDb = $this->sourceClient->selectDatabase($databaseName);
|
||
$targetDb = $this->targetClient->selectDatabase($databaseName);
|
||
|
||
// 获取所有集合
|
||
$collections = $sourceDb->listCollections();
|
||
$batchSize = $this->config['sync']['change_stream']['full_sync_batch_size'] ?? 1000;
|
||
$excludeCollections = $this->config['sync']['exclude_collections'] ?? [];
|
||
|
||
// 统计集合总数，同时预估总文档数和总数据量（用于更精确的进度估算）
|
||
$collectionList = [];
|
||
$totalDocuments = 0;
|
||
$totalBytes = 0;
|
||
|
||
foreach ($collections as $collectionInfo) {
|
||
$collectionName = $collectionInfo->getName();
|
||
if (in_array($collectionName, $excludeCollections)) {
|
||
continue;
|
||
}
|
||
|
||
$collectionList[] = $collectionName;
|
||
|
||
try {
|
||
// 使用 collStats 获取集合的文档数和大小
|
||
$statsCursor = $sourceDb->command(['collStats' => $collectionName]);
|
||
$statsArray = $statsCursor->toArray();
|
||
$collStats = $statsArray[0] ?? [];
|
||
|
||
$collCount = (int)($collStats['count'] ?? 0);
|
||
$collSizeBytes = (int)($collStats['size'] ?? 0);
|
||
|
||
$totalDocuments += $collCount;
|
||
$totalBytes += $collSizeBytes;
|
||
} catch (MongoDBException $e) {
|
||
// 单个集合统计失败不影响整体同步，只记录日志
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_collstats_error',
|
||
'database' => $databaseName,
|
||
'collection' => $collectionName,
|
||
]);
|
||
}
|
||
}
|
||
|
||
// 按名称排序，便于与历史快照稳定对比
|
||
sort($collectionList);
|
||
|
||
// 检测同名数据库结构是否发生重大变化（例如：被删除后重建）
|
||
$collectionsSnapshot = $this->progress['collections_snapshot'] ?? [];
|
||
$previousSnapshot = $collectionsSnapshot[$databaseName] ?? null;
|
||
if ($previousSnapshot !== null && $previousSnapshot !== $collectionList) {
|
||
// 源库结构变化：为了避免旧 checkpoint 导致数据不一致，将该库视为“新库”，重新清空目标并丢弃旧断点
|
||
LoggerHelper::logBusiness('database_sync_source_schema_changed', [
|
||
'database' => $databaseName,
|
||
'previous_collections' => $previousSnapshot,
|
||
'current_collections' => $collectionList,
|
||
], 'warning');
|
||
|
||
// 重新清空目标库
|
||
$this->clearTargetDatabase($databaseName);
|
||
// 丢弃该库的旧断点
|
||
unset($this->progress['checkpoints'][$databaseName]);
|
||
// 标记为已清空
|
||
$clearedDatabases = $this->progress['cleared_databases'] ?? [];
|
||
if (!in_array($databaseName, $clearedDatabases, true)) {
|
||
$clearedDatabases[] = $databaseName;
|
||
}
|
||
$this->progress['cleared_databases'] = $clearedDatabases;
|
||
}
|
||
|
||
// 记录当前集合快照
|
||
$collectionsSnapshot[$databaseName] = $collectionList;
|
||
$this->progress['collections_snapshot'] = $collectionsSnapshot;
|
||
|
||
$this->progress['collections_total'] = count($collectionList);
|
||
$this->progress['collections_completed'] = 0;
|
||
// 为整个数据库预先写入总文档数和总数据量（按库维度估算进度）
|
||
if ($totalDocuments > 0) {
|
||
$this->progress['documents_total'] = $totalDocuments;
|
||
}
|
||
if ($totalBytes > 0) {
|
||
$this->progress['bytes_total'] = $totalBytes;
|
||
}
|
||
// 每次开始全量同步时重置已处理文档数
|
||
$this->progress['documents_processed'] = 0;
|
||
$this->saveProgress();
|
||
|
||
// 根据配置决定是否并行同步集合
|
||
$enableParallel = $this->config['sync']['performance']['enable_parallel_sync'] ?? true;
|
||
$concurrentCollections = $this->config['sync']['performance']['concurrent_collections'] ?? 10;
|
||
|
||
if ($enableParallel && count($collectionList) > 1) {
|
||
// 并行同步多个集合
|
||
$this->syncCollectionsParallel($sourceDb, $targetDb, $collectionList, $databaseName, $batchSize, $concurrentCollections);
|
||
} else {
|
||
// 顺序同步集合
|
||
foreach ($collectionList as $collectionName) {
|
||
$this->syncCollection($sourceDb, $targetDb, $collectionName, $databaseName, $batchSize);
|
||
}
|
||
}
|
||
|
||
$this->stats['databases']++;
|
||
$this->progress['databases_completed']++;
|
||
$this->progress['current_database'] = null;
|
||
$this->progress['current_collection'] = null;
|
||
$this->saveProgress();
|
||
|
||
return true;
|
||
} catch (MongoDBException $e) {
|
||
// 记录错误信息，但不停止整个同步流程
|
||
$errorMessage = $e->getMessage();
|
||
$this->progress['status'] = 'error';
|
||
$this->progress['last_error'] = [
|
||
'message' => $errorMessage,
|
||
'database' => $databaseName,
|
||
'collection' => $this->progress['current_collection'],
|
||
'file' => $e->getFile(),
|
||
'line' => $e->getLine(),
|
||
'time' => date('Y-m-d H:i:s'),
|
||
];
|
||
$this->progress['error_database'] = $databaseName;
|
||
$this->saveProgress();
|
||
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_full_sync_error',
|
||
'database' => $databaseName,
|
||
'collection' => $this->progress['current_collection'],
|
||
]);
|
||
$this->stats['errors']++;
|
||
|
||
// 不返回 false，让调用者决定是否继续同步其他数据库
|
||
// 这样可以跳过有问题的数据库，继续同步其他数据库
|
||
return false;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 同步单个集合（支持大数据量分片）
|
||
*
|
||
* 错误隔离：集合级错误不会影响其他集合的同步
|
||
*/
|
||
private function syncCollection(Database $sourceDb, Database $targetDb, string $collectionName, string $databaseName, int $batchSize): void
|
||
{
|
||
$this->progress['current_collection'] = $collectionName;
|
||
$this->saveProgress();
|
||
|
||
LoggerHelper::logBusiness('database_sync_full_sync_collection_start', [
|
||
'database' => $databaseName,
|
||
'collection' => $collectionName,
|
||
]);
|
||
|
||
try {
|
||
$sourceCollection = $sourceDb->selectCollection($collectionName);
|
||
$targetCollection = $targetDb->selectCollection($collectionName);
|
||
|
||
// 统计当前集合文档总数（用于分片和日志），但不再覆盖全局 documents_total，
|
||
// 全库的总文档数在 fullSyncDatabase 中基于 collStats 预估
|
||
$totalDocuments = $sourceCollection->countDocuments([]);
|
||
|
||
// 检查是否需要分片处理（大数据量）
|
||
$documentsPerTask = $this->config['sync']['performance']['documents_per_task'] ?? 100000;
|
||
$enableParallel = $this->config['sync']['performance']['enable_parallel_sync'] ?? true;
|
||
$maxParallelTasks = $this->config['sync']['performance']['max_parallel_tasks_per_collection'] ?? 4;
|
||
|
||
if ($enableParallel && $totalDocuments > $documentsPerTask && $maxParallelTasks > 1) {
|
||
// 大数据量集合，使用分片并行处理
|
||
$this->syncCollectionParallel($sourceCollection, $targetCollection, $collectionName, $databaseName, $batchSize, $totalDocuments, $maxParallelTasks);
|
||
} else {
|
||
// 小数据量集合，直接同步
|
||
$this->syncCollectionSequential($sourceCollection, $targetCollection, $collectionName, $databaseName, $batchSize);
|
||
}
|
||
|
||
LoggerHelper::logBusiness('database_sync_full_sync_collection_complete', [
|
||
'database' => $databaseName,
|
||
'collection' => $collectionName,
|
||
'count' => $totalDocuments,
|
||
]);
|
||
|
||
$this->stats['collections']++;
|
||
$this->progress['collections_completed']++;
|
||
} catch (\Throwable $e) {
|
||
// 集合级错误隔离：记录错误但继续同步其他集合
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_collection_error',
|
||
'database' => $databaseName,
|
||
'collection' => $collectionName,
|
||
]);
|
||
|
||
$this->stats['errors']++;
|
||
|
||
// 记录集合级错误到进度文件
|
||
$this->progress['last_error'] = [
|
||
'message' => $e->getMessage(),
|
||
'database' => $databaseName,
|
||
'collection' => $collectionName,
|
||
'file' => $e->getFile(),
|
||
'line' => $e->getLine(),
|
||
'time' => date('Y-m-d H:i:s'),
|
||
];
|
||
|
||
// 仍然标记集合为已完成（跳过），继续同步其他集合
|
||
$this->stats['collections']++;
|
||
$this->progress['collections_completed']++;
|
||
} finally {
|
||
$this->progress['current_collection'] = null;
|
||
$this->saveProgress();
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 顺序同步集合（小数据量）
|
||
*/
|
||
private function syncCollectionSequential(Collection $sourceCollection, Collection $targetCollection, string $collectionName, string $databaseName, int $batchSize): void
|
||
{
|
||
// 从断点读取上次同步位置（基于 _id 断点）
|
||
$checkpoint = $this->progress['checkpoints'][$databaseName][$collectionName] ?? null;
|
||
$lastId = $checkpoint['last_id'] ?? null;
|
||
|
||
$filter = [];
|
||
if ($lastId) {
|
||
try {
|
||
$filter['_id'] = ['$gt' => new \MongoDB\BSON\ObjectId($lastId)];
|
||
} catch (\Throwable $e) {
|
||
// 如果 last_id 无法解析为 ObjectId，则退回全量同步
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_invalid_checkpoint_id',
|
||
'database' => $databaseName,
|
||
'collection' => $collectionName,
|
||
'last_id' => $lastId,
|
||
]);
|
||
$filter = [];
|
||
}
|
||
}
|
||
|
||
$options = [
|
||
'batchSize' => $batchSize,
|
||
// 确保按 _id 递增，便于基于 _id 做断点续传
|
||
'sort' => ['_id' => 1],
|
||
];
|
||
|
||
$cursor = $sourceCollection->find($filter, $options);
|
||
$batch = [];
|
||
$lastProgressLogTime = time();
|
||
|
||
foreach ($cursor as $document) {
|
||
$batch[] = $document;
|
||
|
||
if (count($batch) >= $batchSize) {
|
||
$this->batchInsert($targetCollection, $batch);
|
||
$batchCount = count($batch);
|
||
$this->progress['documents_processed'] += $batchCount;
|
||
|
||
// 记录本批次最后一个文档的 _id 作为断点
|
||
$lastDoc = end($batch);
|
||
if (isset($lastDoc['_id'])) {
|
||
$this->progress['checkpoints'][$databaseName][$collectionName] = [
|
||
'last_id' => (string)$lastDoc['_id'],
|
||
'processed' => ($this->progress['checkpoints'][$databaseName][$collectionName]['processed'] ?? 0) + $batchCount,
|
||
'completed' => false,
|
||
];
|
||
}
|
||
|
||
$batch = [];
|
||
|
||
// 每5秒输出一次进度
|
||
if (time() - $lastProgressLogTime >= 5) {
|
||
$this->logProgress();
|
||
$lastProgressLogTime = time();
|
||
}
|
||
$this->saveProgress();
|
||
}
|
||
}
|
||
|
||
// 处理剩余数据
|
||
if (!empty($batch)) {
|
||
$this->batchInsert($targetCollection, $batch);
|
||
$batchCount = count($batch);
|
||
$this->progress['documents_processed'] += $batchCount;
|
||
|
||
$lastDoc = end($batch);
|
||
if (isset($lastDoc['_id'])) {
|
||
$this->progress['checkpoints'][$databaseName][$collectionName] = [
|
||
'last_id' => (string)$lastDoc['_id'],
|
||
'processed' => ($this->progress['checkpoints'][$databaseName][$collectionName]['processed'] ?? 0) + $batchCount,
|
||
'completed' => true,
|
||
];
|
||
}
|
||
|
||
$this->saveProgress();
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 并行同步集合（大数据量，使用分片）
|
||
*/
|
||
private function syncCollectionParallel(Collection $sourceCollection, Collection $targetCollection, string $collectionName, string $databaseName, int $batchSize, int $totalDocuments, int $maxParallelTasks): void
|
||
{
|
||
// 计算每个任务处理的文档数
|
||
$documentsPerTask = (int)ceil($totalDocuments / $maxParallelTasks);
|
||
|
||
LoggerHelper::logBusiness('database_sync_collection_parallel_start', [
|
||
'database' => $databaseName,
|
||
'collection' => $collectionName,
|
||
'total_documents' => $totalDocuments,
|
||
'parallel_tasks' => $maxParallelTasks,
|
||
'documents_per_task' => $documentsPerTask,
|
||
]);
|
||
|
||
// 创建任务列表
|
||
$tasks = [];
|
||
for ($i = 0; $i < $maxParallelTasks; $i++) {
|
||
$skip = $i * $documentsPerTask;
|
||
$limit = min($documentsPerTask, $totalDocuments - $skip);
|
||
|
||
if ($limit <= 0) {
|
||
break;
|
||
}
|
||
|
||
$tasks[] = [
|
||
'skip' => $skip,
|
||
'limit' => $limit,
|
||
'task_id' => $i + 1,
|
||
];
|
||
}
|
||
|
||
// 使用 Workerman 的协程或进程并行执行
|
||
$this->executeParallelTasks($sourceCollection, $targetCollection, $tasks, $batchSize);
|
||
}
|
||
|
||
/**
|
||
* 执行并行任务
|
||
*
|
||
* 注意：由于 Workerman Coroutine 可能存在类加载冲突问题，这里使用顺序执行
|
||
* MongoDB 操作本身已经很快，顺序执行也能保证良好的性能
|
||
*/
|
||
private function executeParallelTasks(Collection $sourceCollection, Collection $targetCollection, array $tasks, int $batchSize): void
|
||
{
|
||
// 顺序执行任务（避免协程类加载冲突）
|
||
foreach ($tasks as $task) {
|
||
$this->syncCollectionChunk($sourceCollection, $targetCollection, $task['skip'], $task['limit'], $batchSize);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 同步集合的一个分片
|
||
*/
|
||
private function syncCollectionChunk(Collection $sourceCollection, Collection $targetCollection, int $skip, int $limit, int $batchSize): void
|
||
{
|
||
$cursor = $sourceCollection->find([], [
|
||
'skip' => $skip,
|
||
'limit' => $limit,
|
||
'batchSize' => $batchSize,
|
||
]);
|
||
|
||
$batch = [];
|
||
$count = 0;
|
||
|
||
foreach ($cursor as $document) {
|
||
$batch[] = $document;
|
||
$count++;
|
||
|
||
if (count($batch) >= $batchSize) {
|
||
$this->batchInsert($targetCollection, $batch);
|
||
$this->progress['documents_processed'] += count($batch);
|
||
$batch = [];
|
||
$this->saveProgress();
|
||
}
|
||
}
|
||
|
||
// 处理剩余数据
|
||
if (!empty($batch)) {
|
||
$this->batchInsert($targetCollection, $batch);
|
||
$this->progress['documents_processed'] += count($batch);
|
||
$this->saveProgress();
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 并行同步多个集合
|
||
*/
|
||
private function syncCollectionsParallel(Database $sourceDb, Database $targetDb, array $collectionList, string $databaseName, int $batchSize, int $concurrentCollections): void
|
||
{
|
||
// 将集合列表分成多个批次
|
||
$chunks = array_chunk($collectionList, $concurrentCollections);
|
||
|
||
foreach ($chunks as $chunk) {
|
||
// 顺序同步当前批次（避免协程类加载冲突）
|
||
// 注意：虽然配置了并发集合数，但由于协程存在类加载问题，这里使用顺序执行
|
||
// MongoDB 操作本身已经很快，顺序执行也能保证良好的性能
|
||
foreach ($chunk as $collectionName) {
|
||
$this->syncCollection($sourceDb, $targetDb, $collectionName, $databaseName, $batchSize);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 批量插入文档（支持重试机制）
|
||
*
|
||
* 错误隔离：文档级错误不会影响其他批次的同步
|
||
*/
|
||
private function batchInsert(Collection $collection, array $documents): void
|
||
{
|
||
if (empty($documents)) {
|
||
return;
|
||
}
|
||
|
||
$maxRetries = $this->config['sync']['retry']['max_sync_retries'] ?? 3;
|
||
$retryDelay = $this->config['sync']['retry']['sync_retry_interval'] ?? 2;
|
||
$retryCount = 0;
|
||
|
||
while ($retryCount <= $maxRetries) {
|
||
try {
|
||
// 使用 bulkWrite 进行批量写入
|
||
$operations = [];
|
||
foreach ($documents as $doc) {
|
||
$operations[] = [
|
||
'insertOne' => [$doc],
|
||
];
|
||
}
|
||
|
||
$collection->bulkWrite($operations, ['ordered' => false]);
|
||
$this->stats['documents_inserted'] += count($documents);
|
||
return; // 成功，退出重试循环
|
||
} catch (MongoDBException $e) {
|
||
$retryCount++;
|
||
|
||
if ($retryCount > $maxRetries) {
|
||
// 超过最大重试次数，记录错误但继续处理下一批
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_batch_insert_error',
|
||
'collection' => $collection->getCollectionName(),
|
||
'count' => count($documents),
|
||
'retry_count' => $retryCount - 1,
|
||
]);
|
||
|
||
$this->stats['errors']++;
|
||
|
||
// 对于文档级错误，不抛出异常，继续处理下一批
|
||
// 这样可以保证即使某些文档失败，也能继续同步其他文档
|
||
return;
|
||
}
|
||
|
||
// 指数退避重试
|
||
$delay = $retryDelay * pow(2, $retryCount - 1);
|
||
LoggerHelper::logBusiness('database_sync_batch_insert_retry', [
|
||
'collection' => $collection->getCollectionName(),
|
||
'retry_count' => $retryCount,
|
||
'max_retries' => $maxRetries,
|
||
'delay' => $delay,
|
||
]);
|
||
|
||
// 等待后重试
|
||
sleep($delay);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 监听数据库变化并同步
|
||
*
|
||
* 注意：此方法会阻塞，需要在独立进程中运行
|
||
*/
|
||
public function watchDatabase(string $databaseName): void
|
||
{
|
||
try {
|
||
// 确保目标数据库存在
|
||
$this->ensureTargetDatabaseExists($databaseName);
|
||
|
||
$sourceDb = $this->sourceClient->selectDatabase($databaseName);
|
||
$targetDb = $this->targetClient->selectDatabase($databaseName);
|
||
|
||
$batchSize = $this->config['sync']['change_stream']['batch_size'] ?? 100;
|
||
$maxAwaitTimeMs = $this->config['sync']['change_stream']['max_await_time_ms'] ?? 1000;
|
||
$excludeCollections = $this->config['sync']['exclude_collections'] ?? [];
|
||
|
||
// 使用数据库级别的 Change Stream（MongoDB 4.0+）
|
||
// 这样可以监听整个数据库的所有集合变化
|
||
$changeStream = $sourceDb->watch(
|
||
[],
|
||
[
|
||
'fullDocument' => 'updateLookup',
|
||
'batchSize' => $batchSize,
|
||
'maxAwaitTimeMS' => $maxAwaitTimeMs,
|
||
]
|
||
);
|
||
|
||
LoggerHelper::logBusiness('database_sync_watch_database_start', [
|
||
'database' => $databaseName,
|
||
]);
|
||
|
||
// 处理变更事件
|
||
foreach ($changeStream as $change) {
|
||
$collectionName = $change['ns']['coll'] ?? '';
|
||
|
||
// 排除系统集合
|
||
if (in_array($collectionName, $excludeCollections)) {
|
||
continue;
|
||
}
|
||
|
||
// 获取目标集合
|
||
$targetCollection = $targetDb->selectCollection($collectionName);
|
||
|
||
// 处理变更
|
||
$this->processChange($targetCollection, $change);
|
||
$this->stats['last_sync_time'] = time();
|
||
$this->progress['status'] = 'incremental_sync';
|
||
$this->saveProgress();
|
||
}
|
||
} catch (MongoDBException $e) {
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_watch_database_error',
|
||
'database' => $databaseName,
|
||
]);
|
||
throw $e;
|
||
}
|
||
}
|
||
|
||
|
||
/**
|
||
* 处理变更事件
|
||
*/
|
||
private function processChange(Collection $targetCollection, $change): void
|
||
{
|
||
try {
|
||
$operationType = $change['operationType'] ?? '';
|
||
|
||
switch ($operationType) {
|
||
case 'insert':
|
||
$this->handleInsert($targetCollection, $change);
|
||
break;
|
||
case 'update':
|
||
case 'replace':
|
||
$this->handleUpdate($targetCollection, $change);
|
||
break;
|
||
case 'delete':
|
||
$this->handleDelete($targetCollection, $change);
|
||
break;
|
||
default:
|
||
LoggerHelper::logBusiness('database_sync_unknown_operation', [
|
||
'operation' => $operationType,
|
||
'collection' => $targetCollection->getCollectionName(),
|
||
]);
|
||
}
|
||
} catch (MongoDBException $e) {
|
||
LoggerHelper::logError($e, [
|
||
'action' => 'database_sync_process_change_error',
|
||
'collection' => $targetCollection->getCollectionName(),
|
||
'operation' => $change['operationType'] ?? 'unknown',
|
||
]);
|
||
$this->stats['errors']++;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 处理插入操作
|
||
*/
|
||
private function handleInsert(Collection $targetCollection, array $change): void
|
||
{
|
||
$document = $change['fullDocument'] ?? null;
|
||
if ($document) {
|
||
$targetCollection->insertOne($document);
|
||
$this->stats['documents_inserted']++;
|
||
|
||
if ($this->config['monitoring']['log_detail'] ?? false) {
|
||
LoggerHelper::logBusiness('database_sync_insert', [
|
||
'collection' => $targetCollection->getCollectionName(),
|
||
'document_id' => (string)($document['_id'] ?? ''),
|
||
]);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 处理更新操作
|
||
*/
|
||
private function handleUpdate(Collection $targetCollection, array $change): void
|
||
{
|
||
$documentId = $change['documentKey']['_id'] ?? null;
|
||
$fullDocument = $change['fullDocument'] ?? null;
|
||
|
||
if ($documentId) {
|
||
if ($fullDocument) {
|
||
// 使用完整文档替换
|
||
$targetCollection->replaceOne(
|
||
['_id' => $documentId],
|
||
$fullDocument,
|
||
['upsert' => true]
|
||
);
|
||
} else {
|
||
// 使用更新操作
|
||
$updateDescription = $change['updateDescription'] ?? [];
|
||
$updatedFields = $updateDescription['updatedFields'] ?? [];
|
||
$removedFields = $updateDescription['removedFields'] ?? [];
|
||
|
||
$update = [];
|
||
if (!empty($updatedFields)) {
|
||
$update['$set'] = $updatedFields;
|
||
}
|
||
if (!empty($removedFields)) {
|
||
$update['$unset'] = array_fill_keys($removedFields, '');
|
||
}
|
||
|
||
if (!empty($update)) {
|
||
$targetCollection->updateOne(
|
||
['_id' => $documentId],
|
||
$update,
|
||
['upsert' => true]
|
||
);
|
||
}
|
||
}
|
||
$this->stats['documents_updated']++;
|
||
|
||
if ($this->config['monitoring']['log_detail'] ?? false) {
|
||
LoggerHelper::logBusiness('database_sync_update', [
|
||
'collection' => $targetCollection->getCollectionName(),
|
||
'document_id' => (string)$documentId,
|
||
]);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 处理删除操作
|
||
*/
|
||
private function handleDelete(Collection $targetCollection, array $change): void
|
||
{
|
||
$documentId = $change['documentKey']['_id'] ?? null;
|
||
if ($documentId) {
|
||
$targetCollection->deleteOne(['_id' => $documentId]);
|
||
$this->stats['documents_deleted']++;
|
||
|
||
if ($this->config['monitoring']['log_detail'] ?? false) {
|
||
LoggerHelper::logBusiness('database_sync_delete', [
|
||
'collection' => $targetCollection->getCollectionName(),
|
||
'document_id' => (string)$documentId,
|
||
]);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 获取同步统计信息
|
||
*/
|
||
public function getStats(): array
|
||
{
|
||
return $this->stats;
|
||
}
|
||
|
||
/**
|
||
* 获取同步进度信息
|
||
*/
|
||
public function getProgress(): array
|
||
{
|
||
// 计算进度百分比（优先使用文档级进度，更准确）
|
||
$progressPercent = 0;
|
||
|
||
// 方法1：基于文档数计算（最准确）
|
||
if ($this->progress['documents_total'] > 0 && $this->progress['documents_processed'] > 0) {
|
||
$docProgress = ($this->progress['documents_processed'] / $this->progress['documents_total']) * 100;
|
||
$progressPercent = round($docProgress, 2);
|
||
}
|
||
// 方法2：基于数据库和集合计算（备用）
|
||
elseif ($this->progress['databases_total'] > 0) {
|
||
$dbProgress = ($this->progress['databases_completed'] / $this->progress['databases_total']) * 100;
|
||
|
||
// 如果当前正在处理某个数据库，考虑集合进度
|
||
if ($this->progress['collections_total'] > 0 && $this->progress['current_database']) {
|
||
$collectionProgress = ($this->progress['collections_completed'] / $this->progress['collections_total']) * 100;
|
||
// 当前数据库的进度 = 已完成数据库数 + 当前数据库的集合进度
|
||
$dbProgress = ($this->progress['databases_completed'] + ($collectionProgress / 100)) / $this->progress['databases_total'] * 100;
|
||
}
|
||
|
||
$progressPercent = round($dbProgress, 2);
|
||
}
|
||
|
||
// 确保进度在 0-100 之间
|
||
$progressPercent = max(0, min(100, $progressPercent));
|
||
|
||
// 计算已用时间
|
||
$elapsedTime = null;
|
||
if ($this->progress['start_time']) {
|
||
$elapsedTime = round(microtime(true) - $this->progress['start_time'], 2);
|
||
}
|
||
|
||
// 基于文档数和预估总数据量，计算按“数据量”的同步进度（字节级）
|
||
$bytesTotal = (int)($this->progress['bytes_total'] ?? 0);
|
||
$bytesProcessed = 0;
|
||
if ($bytesTotal > 0 && $this->progress['documents_total'] > 0) {
|
||
$ratio = $this->progress['documents_processed'] / max(1, $this->progress['documents_total']);
|
||
if ($ratio > 1) {
|
||
$ratio = 1;
|
||
} elseif ($ratio < 0) {
|
||
$ratio = 0;
|
||
}
|
||
$bytesProcessed = (int)round($bytesTotal * $ratio);
|
||
}
|
||
|
||
// 计算预计剩余时间
|
||
$estimatedRemaining = null;
|
||
if ($progressPercent > 0 && $elapsedTime) {
|
||
$totalEstimatedTime = $elapsedTime / ($progressPercent / 100);
|
||
$estimatedRemaining = round($totalEstimatedTime - $elapsedTime, 2);
|
||
}
|
||
|
||
return [
|
||
'status' => $this->progress['status'],
|
||
'progress_percent' => $progressPercent,
|
||
'current_database' => $this->progress['current_database'],
|
||
'current_collection' => $this->progress['current_collection'],
|
||
'databases' => [
|
||
'total' => $this->progress['databases_total'],
|
||
'completed' => $this->progress['databases_completed'],
|
||
'remaining' => $this->progress['databases_total'] - $this->progress['databases_completed'],
|
||
],
|
||
'collections' => [
|
||
'total' => $this->progress['collections_total'],
|
||
'completed' => $this->progress['collections_completed'],
|
||
'remaining' => $this->progress['collections_total'] - $this->progress['collections_completed'],
|
||
],
|
||
'documents' => [
|
||
'total' => $this->progress['documents_total'],
|
||
'processed' => $this->progress['documents_processed'],
|
||
'remaining' => max(0, $this->progress['documents_total'] - $this->progress['documents_processed']),
|
||
],
|
||
'bytes' => [
|
||
'total' => $bytesTotal,
|
||
'processed' => $bytesProcessed,
|
||
'remaining' => max(0, $bytesTotal - $bytesProcessed),
|
||
],
|
||
'time' => [
|
||
'elapsed_seconds' => $elapsedTime,
|
||
'estimated_remaining_seconds' => $estimatedRemaining,
|
||
'start_time' => $this->progress['start_time'] ? date('Y-m-d H:i:s', (int)$this->progress['start_time']) : null,
|
||
],
|
||
'stats' => $this->stats,
|
||
'last_error' => $this->progress['last_error'] ?? null,
|
||
'error_database' => $this->progress['error_database'] ?? null,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 重置进度并清除错误状态（用于恢复同步）
|
||
*/
|
||
public function resetProgress(): void
|
||
{
|
||
$this->resetStats();
|
||
LoggerHelper::logBusiness('database_sync_progress_reset', []);
|
||
}
|
||
|
||
/**
|
||
* 跳过当前错误数据库，继续同步下一个
|
||
*/
|
||
public function skipErrorDatabase(): bool
|
||
{
|
||
if ($this->progress['status'] === 'error' && $this->progress['error_database']) {
|
||
$errorDb = $this->progress['error_database'];
|
||
|
||
// 标记该数据库为已完成（跳过）
|
||
$this->stats['databases']++;
|
||
$this->progress['databases_completed']++;
|
||
|
||
// 清除错误状态
|
||
$this->progress['status'] = 'full_sync';
|
||
$this->progress['current_database'] = null;
|
||
$this->progress['current_collection'] = null;
|
||
$this->progress['error_database'] = null;
|
||
$this->progress['last_error'] = null;
|
||
|
||
$this->saveProgress();
|
||
|
||
LoggerHelper::logBusiness('database_sync_skip_error_database', [
|
||
'database' => $errorDb,
|
||
]);
|
||
|
||
return true;
|
||
}
|
||
return false;
|
||
}
|
||
|
||
/**
|
||
* 获取运行时目录路径
|
||
*/
|
||
private function getRuntimePath(): string
|
||
{
|
||
if (function_exists('runtime_path')) {
|
||
$path = runtime_path();
|
||
} else {
|
||
$basePath = function_exists('base_path') ? base_path() : __DIR__ . '/../../';
|
||
$path = config('app.runtime_path', $basePath . DIRECTORY_SEPARATOR . 'runtime');
|
||
}
|
||
if (!is_dir($path)) {
|
||
mkdir($path, 0777, true);
|
||
}
|
||
return $path;
|
||
}
|
||
|
||
/**
|
||
* 保存进度到文件（用于多进程共享）
|
||
*
|
||
* 使用文件锁（LOCK_EX）保证多进程写入的原子性，避免并发冲突
|
||
*/
|
||
private function saveProgress(): void
|
||
{
|
||
$progressFile = $this->getRuntimePath() . DIRECTORY_SEPARATOR . 'database_sync_progress.json';
|
||
|
||
// 使用文件锁保证原子性写入
|
||
$fp = fopen($progressFile, 'c+'); // 'c+' 模式：如果文件不存在则创建，如果存在则打开用于读写
|
||
if ($fp === false) {
|
||
LoggerHelper::logError(new \RuntimeException("无法打开进度文件: {$progressFile}"), [
|
||
'action' => 'database_sync_save_progress_error',
|
||
]);
|
||
return;
|
||
}
|
||
|
||
// 获取独占锁（LOCK_EX），阻塞直到获取锁
|
||
if (flock($fp, LOCK_EX)) {
|
||
try {
|
||
// 读取现有进度（如果存在），智能合并更新
|
||
$existingContent = stream_get_contents($fp);
|
||
if ($existingContent) {
|
||
$existingProgress = json_decode($existingContent, true);
|
||
if ($existingProgress && is_array($existingProgress)) {
|
||
// 智能合并策略：
|
||
// 1. 保留全局统计信息（databases_total, collections_total 等）
|
||
// 2. 合并 checkpoints（每个进程只更新自己负责的数据库）
|
||
// 3. 合并 cleared_databases（避免重复清空）
|
||
// 4. 合并 collections_snapshot（保留所有数据库的快照）
|
||
// 5. 更新当前进程的进度信息
|
||
|
||
// 保留全局统计（取最大值，确保不丢失）
|
||
$this->progress['databases_total'] = max(
|
||
$this->progress['databases_total'] ?? 0,
|
||
$existingProgress['databases_total'] ?? 0
|
||
);
|
||
$this->progress['collections_total'] = max(
|
||
$this->progress['collections_total'] ?? 0,
|
||
$existingProgress['collections_total'] ?? 0
|
||
);
|
||
$this->progress['documents_total'] = max(
|
||
$this->progress['documents_total'] ?? 0,
|
||
$existingProgress['documents_total'] ?? 0
|
||
);
|
||
$this->progress['bytes_total'] = max(
|
||
$this->progress['bytes_total'] ?? 0,
|
||
$existingProgress['bytes_total'] ?? 0
|
||
);
|
||
|
||
// 对于 completed 计数，需要累加（多进程场景）
|
||
// 但由于每个进程只处理部分数据库，直接累加会导致重复计数
|
||
// 因此采用基于 checkpoints 重新计算的方式
|
||
// 这里先保留现有值，在 getProgress 中基于 checkpoints 重新计算
|
||
// 为了简化，这里采用取最大值的方式（每个进程只更新自己完成的部分）
|
||
// 注意：这种方式在多进程场景下可能不够精确，但可以避免重复计数
|
||
$this->progress['databases_completed'] = max(
|
||
$this->progress['databases_completed'] ?? 0,
|
||
$existingProgress['databases_completed'] ?? 0
|
||
);
|
||
$this->progress['collections_completed'] = max(
|
||
$this->progress['collections_completed'] ?? 0,
|
||
$existingProgress['collections_completed'] ?? 0
|
||
);
|
||
$this->progress['documents_processed'] = max(
|
||
$this->progress['documents_processed'] ?? 0,
|
||
$existingProgress['documents_processed'] ?? 0
|
||
);
|
||
|
||
// 合并 checkpoints（每个进程只更新自己负责的数据库）
|
||
if (isset($existingProgress['checkpoints']) && is_array($existingProgress['checkpoints'])) {
|
||
if (!isset($this->progress['checkpoints'])) {
|
||
$this->progress['checkpoints'] = [];
|
||
}
|
||
$this->progress['checkpoints'] = array_merge(
|
||
$existingProgress['checkpoints'],
|
||
$this->progress['checkpoints']
|
||
);
|
||
}
|
||
|
||
// 合并 cleared_databases（避免重复清空）
|
||
if (isset($existingProgress['cleared_databases']) && is_array($existingProgress['cleared_databases'])) {
|
||
if (!isset($this->progress['cleared_databases'])) {
|
||
$this->progress['cleared_databases'] = [];
|
||
}
|
||
$this->progress['cleared_databases'] = array_unique(array_merge(
|
||
$existingProgress['cleared_databases'],
|
||
$this->progress['cleared_databases']
|
||
));
|
||
}
|
||
|
||
// 合并 collections_snapshot（保留所有数据库的快照）
|
||
if (isset($existingProgress['collections_snapshot']) && is_array($existingProgress['collections_snapshot'])) {
|
||
if (!isset($this->progress['collections_snapshot'])) {
|
||
$this->progress['collections_snapshot'] = [];
|
||
}
|
||
$this->progress['collections_snapshot'] = array_merge(
|
||
$existingProgress['collections_snapshot'],
|
||
$this->progress['collections_snapshot']
|
||
);
|
||
}
|
||
|
||
// 合并 orphan_databases
|
||
if (isset($existingProgress['orphan_databases']) && is_array($existingProgress['orphan_databases'])) {
|
||
if (!isset($this->progress['orphan_databases'])) {
|
||
$this->progress['orphan_databases'] = [];
|
||
}
|
||
$this->progress['orphan_databases'] = array_unique(array_merge(
|
||
$existingProgress['orphan_databases'],
|
||
$this->progress['orphan_databases']
|
||
));
|
||
}
|
||
|
||
// 保留最早的 start_time
|
||
if (isset($existingProgress['start_time']) && $existingProgress['start_time'] > 0) {
|
||
if (!isset($this->progress['start_time']) || $this->progress['start_time'] === null) {
|
||
$this->progress['start_time'] = $existingProgress['start_time'];
|
||
} else {
|
||
$this->progress['start_time'] = min(
|
||
$this->progress['start_time'],
|
||
$existingProgress['start_time']
|
||
);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// 清空文件并写入合并后的进度
|
||
ftruncate($fp, 0);
|
||
rewind($fp);
|
||
fwrite($fp, json_encode($this->progress, JSON_PRETTY_PRINT | JSON_UNESCAPED_UNICODE));
|
||
fflush($fp); // 确保立即写入磁盘
|
||
} finally {
|
||
// 释放锁
|
||
flock($fp, LOCK_UN);
|
||
}
|
||
} else {
|
||
LoggerHelper::logError(new \RuntimeException("无法获取进度文件锁: {$progressFile}"), [
|
||
'action' => 'database_sync_save_progress_lock_error',
|
||
]);
|
||
}
|
||
|
||
fclose($fp);
|
||
}
|
||
|
||
/**
|
||
* 设置进度状态（公开方法，供外部调用）
|
||
*/
|
||
public function setProgressStatus(string $status): void
|
||
{
|
||
$this->progress['status'] = $status;
|
||
if ($status !== 'idle' && $this->progress['start_time'] === null) {
|
||
$this->progress['start_time'] = microtime(true);
|
||
}
|
||
$this->saveProgress();
|
||
}
|
||
|
||
/**
|
||
* 从文件加载进度（使用文件锁保证读取一致性）
|
||
*/
|
||
public function loadProgress(): void
|
||
{
|
||
$progressFile = $this->getRuntimePath() . DIRECTORY_SEPARATOR . 'database_sync_progress.json';
|
||
if (!file_exists($progressFile)) {
|
||
return;
|
||
}
|
||
|
||
// 使用文件锁保证读取一致性
|
||
$fp = fopen($progressFile, 'r');
|
||
if ($fp === false) {
|
||
LoggerHelper::logError(new \RuntimeException("无法打开进度文件: {$progressFile}"), [
|
||
'action' => 'database_sync_load_progress_error',
|
||
]);
|
||
return;
|
||
}
|
||
|
||
// 获取共享锁（LOCK_SH），允许多个进程同时读取
|
||
if (flock($fp, LOCK_SH)) {
|
||
try {
|
||
$content = stream_get_contents($fp);
|
||
if ($content) {
|
||
$loaded = json_decode($content, true);
|
||
if ($loaded && is_array($loaded)) {
|
||
// 合并进度：保留现有字段，更新加载的字段
|
||
$this->progress = array_merge($this->progress, $loaded);
|
||
}
|
||
}
|
||
} finally {
|
||
// 释放锁
|
||
flock($fp, LOCK_UN);
|
||
}
|
||
} else {
|
||
LoggerHelper::logError(new \RuntimeException("无法获取进度文件锁: {$progressFile}"), [
|
||
'action' => 'database_sync_load_progress_lock_error',
|
||
]);
|
||
}
|
||
|
||
fclose($fp);
|
||
}
|
||
|
||
/**
|
||
* 输出进度日志
|
||
*/
|
||
private function logProgress(): void
|
||
{
|
||
$progress = $this->getProgress();
|
||
LoggerHelper::logBusiness('database_sync_progress', [
|
||
'status' => $progress['status'],
|
||
'progress_percent' => $progress['progress_percent'] . '%',
|
||
'current_database' => $progress['current_database'],
|
||
'current_collection' => $progress['current_collection'],
|
||
'databases' => "{$progress['databases']['completed']}/{$progress['databases']['total']}",
|
||
'collections' => "{$progress['collections']['completed']}/{$progress['collections']['total']}",
|
||
'documents' => "{$progress['documents']['processed']}/{$progress['documents']['total']}",
|
||
'elapsed_time' => $progress['time']['elapsed_seconds'] . 's',
|
||
'estimated_remaining' => $progress['time']['estimated_remaining_seconds'] ? $progress['time']['estimated_remaining_seconds'] . 's' : 'calculating...',
|
||
]);
|
||
}
|
||
|
||
/**
|
||
* 重置统计信息
|
||
*/
|
||
public function resetStats(): void
|
||
{
|
||
$this->stats = [
|
||
'databases' => 0,
|
||
'collections' => 0,
|
||
'documents_inserted' => 0,
|
||
'documents_updated' => 0,
|
||
'documents_deleted' => 0,
|
||
'errors' => 0,
|
||
'last_sync_time' => null,
|
||
];
|
||
$this->progress = [
|
||
'status' => 'idle',
|
||
'current_database' => null,
|
||
'current_collection' => null,
|
||
'databases_total' => 0,
|
||
'databases_completed' => 0,
|
||
'collections_total' => 0,
|
||
'collections_completed' => 0,
|
||
'documents_total' => 0,
|
||
'documents_processed' => 0,
|
||
'bytes_total' => 0,
|
||
'cleared_databases' => [],
|
||
'collections_snapshot' => [],
|
||
'orphan_databases' => [],
|
||
'start_time' => null,
|
||
'current_database_start_time' => null,
|
||
'estimated_time_remaining' => null,
|
||
'last_error' => null,
|
||
'error_database' => null,
|
||
];
|
||
$this->saveProgress();
|
||
}
|
||
}
|
||
|