WuYongJie/CaiWuXiTong
1
0
Fork 0
Code Issues Pull Requests 2 Packages Projects Releases Wiki Activity

优化订单与账单镜像分批读取及内存释放

Browse Source
This commit is contained in:
WuYongJie 2026-10-19 21:04:42 +08:00
parent bbb967e110
commit 31f6c72bf3
3 changed files with 171 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
config/upstream_mirrors.php
View File

@ -193,6 +193,13 @@ return [
*/
'table_sync' => [
'batch_size' => (int) env('BUSINESS_MIRROR_TABLE_BATCH_SIZE', 500),
// These wide SQL Server views time out when TOP/pagination is planned
// together with every payload column. Select the bounded keys first,
// then look up only those rows in the same read-only statement.
'key_first_datasets' => [
'operations_replica.rpt.jst_Orders',
'operations_replica.rpt.jst_originalBillInfo',
],
'max_batches_per_run' => (int) env('BUSINESS_MIRROR_TABLE_MAX_BATCHES', 2),
'max_datasets_per_run' => (int) env('BUSINESS_MIRROR_MAX_DATASETS', 1),
'overlap_hours' => (int) env('BUSINESS_MIRROR_OVERLAP_HOURS', 48),

70
modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php
View File

@ -3,9 +3,11 @@
namespace Modules\EcommerceFinance\Services;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Str;
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
use PDOException;
use RuntimeException;
use Throwable;
@ -64,9 +66,12 @@ class BusinessMirrorSynchronizer
$batches = 0;
$deadline = microtime(true) + max((int) config('upstream_pipeline.sync.dataset_seconds', 8), 1);
$rows = (int) ($state['last_upserted_rows'] ?? 0);
$started = microtime(true);
$stage = 'read_upstream';
try {
while (true) {
$stage = 'read_upstream';
$batch = $this->readBatch(
$dataset,
$columns,
@ -79,17 +84,24 @@ class BusinessMirrorSynchronizer
);
if ($batch === []) {
$stage = 'complete_mirror';
$this->mirror->complete($dataset, $mode, $generation, $cursor ?? $highWatermark, $rows);
return $this->result($dataset, 'succeeded', $mode, $batches, $rows);
}
$stage = 'normalize_rows';
$cursor = $this->cursor($dataset, $batch[array_key_last($batch)]);
$rows += count($batch);
$batchRows = count($batch);
$rows += $batchRows;
$batches++;
$this->mirror->storeBatch($dataset, $columns, $this->normalize($columns, $batch), $cursor, $generation, $rows);
$batch = $this->normalize($columns, $batch);
$stage = 'write_mirror';
$this->mirror->storeBatch($dataset, $columns, $batch, $cursor, $generation, $rows);
unset($batch);
if (count($batch) < $batchSize) {
if ($batchRows < $batchSize) {
$stage = 'complete_mirror';
$this->mirror->complete($dataset, $mode, $generation, $cursor, $rows);
return $this->result($dataset, 'succeeded', $mode, $batches, $rows);
@ -100,6 +112,28 @@ class BusinessMirrorSynchronizer
}
}
} catch (Throwable $exception) {
// PDOException alone cannot distinguish an upstream timeout from a
// local transaction failure. Keep the stage and numeric diagnostics,
// never the exception message, SQL, bindings or connection details.
$sqlState = null;
$driverCode = null;
for ($cause = $exception; $cause !== null; $cause = $cause->getPrevious()) {
if ($cause instanceof PDOException) {
$code = (string) ($cause->errorInfo[0] ?? $cause->getCode());
$sqlState = preg_match('/^[A-Z0-9]{5}$/D', $code) === 1 ? $code : null;
$native = $cause->errorInfo[1] ?? null;
$driverCode = is_numeric($native) ? (int) $native : null;
}
}
Log::warning('共享业务镜像同步阶段失败。', [
'dataset' => $dataset->key(),
'source' => $dataset->sourceKey(),
'stage' => $stage,
'sql_state' => $sqlState,
'driver_code' => $driverCode,
'elapsed_seconds' => round(microtime(true) - $started, 3),
'exception' => $exception::class,
]);
$this->mirror->fail($dataset, $exception::class);
throw $exception;
}
@ -157,33 +191,49 @@ class BusinessMirrorSynchronizer
implode(', ', array_map(fn (array $column): string => $this->selectExpression($column), $columns)), $tail, $limit,
);
if ($this->driver === 'sqlsrv' && $mode === 'incremental'
&& in_array($dataset->key(), (array) config('upstream_mirrors.table_sync.key_first_datasets', []), true)) {
$keys = (new UpstreamSqlDialect($this->driver))->select(
implode(', ', array_map(fn (string $name): string => $this->quote($name), $order)), $tail, $limit,
);
$keyColumn = fn (string $name): string => $this->quote('mirror_keys') . '.' . $this->quote($name);
$rowColumn = fn (string $name): string => $this->quote('mirror_rows') . '.' . $this->quote($name);
$sql = 'SELECT ' . implode(', ', array_map(fn (array $column): string => $this->selectExpression($column, 'mirror_rows'), $columns))
. ' FROM (' . $keys . ') AS ' . $this->quote('mirror_keys')
. ' INNER LOOP JOIN ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject())
. ' AS ' . $this->quote('mirror_rows')
. ' ON ' . implode(' AND ', array_map(fn (string $name): string => $rowColumn($name) . ' = ' . $keyColumn($name), $order))
. ' ORDER BY ' . implode(', ', array_map($keyColumn, $order));
}
return $this->queries->forSource($dataset->sourceKey())->select($sql, $bindings);
}
/** @param array<string,mixed> $column */
private function selectExpression(array $column): string
private function selectExpression(array $column, ?string $alias = null): string
{
$name = (string) $column['column_name'];
$quoted = $this->quote($name);
$reference = $alias === null ? $quoted : $this->quote($alias) . '.' . $quoted;
if ($this->driver !== 'sqlsrv') {
if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'timestamp with time zone') {
// Store timezone-aware PostgreSQL instants as UTC in the shared mirror.
return "to_char({$quoted} AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') AS {$quoted}";
return "to_char({$reference} AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') AS {$quoted}";
}
if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'bytea') {
return "encode({$quoted}, 'hex') AS {$quoted}";
return "encode({$reference}, 'hex') AS {$quoted}";
}
return $quoted;
return $reference;
}
// FreeTDS may return SQL Server date values using the connection locale
// (for example "Jan 1 2026"). Convert temporal columns at the source so
// every mirror receives deterministic ISO values that MariaDB accepts.
return match (strtolower((string) ($column['data_type'] ?? ''))) {
'date' => 'CONVERT(varchar(10), ' . $quoted . ', 23) AS ' . $quoted,
'datetime', 'datetime2', 'smalldatetime' => 'CONVERT(varchar(27), ' . $quoted . ', 126) AS ' . $quoted,
default => $quoted,
'date' => 'CONVERT(varchar(10), ' . $reference . ', 23) AS ' . $quoted,
'datetime', 'datetime2', 'smalldatetime' => 'CONVERT(varchar(27), ' . $reference . ', 126) AS ' . $quoted,
default => $reference,
};
}

106
tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php
View File

@ -4,6 +4,7 @@ namespace Tests\Unit\EcommerceFinance;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
@ -145,6 +146,37 @@ class BusinessMirrorSynchronizerTest extends TestCase
}
}
public function test_key_first_incremental_read_preserves_filters_projection_and_resume_keys(): void
{
$this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '旧值']]];
$this->synchronizer->sync('test.dbo.records', true, 10, 1);
config(['upstream_mirrors.table_sync.key_first_datasets' => ['test.dbo.records']]);
$this->query->calls = [];
$this->query->batches = [
[['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => '更新']],
[['Id' => 2, 'updateTime' => '2026-08-15 10:01:00', 'displayName' => '新增']],
[],
];
$this->assertSame('running', $this->synchronizer->sync('test.dbo.records', false, 1, 1)['status']);
$this->assertSame('succeeded', $this->synchronizer->sync('test.dbo.records', false, 1, 3)['status']);
$sql = $this->query->calls[0]['sql'];
$this->assertStringContainsString('FROM (SELECT TOP (1) [Id] FROM [dbo].[records] WHERE ([updateTime] >= ? AND [updateTime] < ?) ORDER BY [Id]) AS [mirror_keys]', $sql);
$this->assertStringContainsString('INNER LOOP JOIN [dbo].[records] AS [mirror_rows] ON [mirror_rows].[Id] = [mirror_keys].[Id]', $sql);
$this->assertStringContainsString('CONVERT(varchar(27), [mirror_rows].[updateTime], 126) AS [updateTime]', $sql);
$this->assertStringContainsString('ORDER BY [mirror_keys].[Id]', $sql);
$this->assertCount(2, $this->query->calls[0]['bindings']);
$this->assertSame(1, $this->query->calls[1]['bindings'][2]);
UpstreamReadOnlyQuery::assertReadOnlySql($sql);
$this->assertSame(2, DB::connection('business_mirror')->table('test_records')->count());
$this->assertSame('更新', DB::connection('business_mirror')->table('test_records')->where('Id', 1)->value('displayName'));
$this->query->calls = [];
$this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => '更新']]];
$this->synchronizer->sync('test.dbo.records', true, 10, 1);
$this->assertStringNotContainsString('LOOP JOIN', $this->query->calls[0]['sql']);
}
public function test_fixed_filters_are_validated_and_bound_to_every_source_batch(): void
{
DB::table('upstream_schema_columns')->insert($this->column('channel', 5, 'nvarchar', 50));
@ -507,6 +539,71 @@ class BusinessMirrorSynchronizerTest extends TestCase
);
}
public function test_upstream_timeout_keeps_previous_data_and_logs_only_safe_diagnostics(): void
{
$this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '保留']]];
$this->synchronizer->sync('test.dbo.records', true, 10, 1);
$previousSuccess = DB::connection('business_mirror')->table('business_mirror_table_states')->value('last_succeeded_at');
$exception = new \PDOException('secret DSN and financial SQL must never be logged');
$exception->errorInfo = ['HY000', 20003, 'secret native message'];
$this->query->batches = [$exception];
Log::spy();
try {
$this->synchronizer->sync('test.dbo.records', false, 10, 1);
$this->fail('Expected the upstream failure to propagate.');
} catch (\PDOException $caught) {
$this->assertSame($exception, $caught);
}
Log::shouldHaveReceived('warning')->once()->withArgs(function (string $message, array $context): bool {
$this->assertSame('read_upstream', $context['stage']);
$this->assertSame('HY000', $context['sql_state']);
$this->assertSame(20003, $context['driver_code']);
$this->assertStringNotContainsString('secret', json_encode($context));
$this->assertArrayNotHasKey('message', $context);
$this->assertArrayNotHasKey('sql', $context);
return true;
});
$state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
$this->assertSame('failed', $state->status);
$this->assertSame($previousSuccess, $state->last_succeeded_at);
$this->assertSame('保留', DB::connection('business_mirror')->table('test_records')->value('displayName'));
}
public function test_local_pdo_failure_is_identified_as_a_mirror_write(): void
{
$this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '保留']]];
$this->synchronizer->sync('test.dbo.records', true, 10, 1);
$exception = new \PDOException('private local database error');
$exception->errorInfo = ['40001', 1213, 'private message'];
$mirror = \Mockery::mock(GenericBusinessMirrorRepository::class, [app('db')])->makePartial();
$mirror->shouldReceive('storeBatch')->once()->andThrow($exception);
$factory = new class($this->query) extends UpstreamReadOnlyQueryFactory {
public function __construct(private readonly UpstreamReadOnlyQuery $query) {}
public function forSource(string $source): UpstreamReadOnlyQuery { return $this->query; }
};
$synchronizer = new BusinessMirrorSynchronizer(new BusinessMirrorDatasetRegistry(), new BusinessMirrorSchemaManager(), $mirror, $factory);
$this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => '未写入']]];
Log::spy();
try {
$synchronizer->sync('test.dbo.records', false, 10, 1);
$this->fail('Expected the local failure to propagate.');
} catch (\PDOException $caught) {
$this->assertSame($exception, $caught);
}
Log::shouldHaveReceived('warning')->once()->withArgs(function (string $message, array $context): bool {
$this->assertSame('write_mirror', $context['stage']);
$this->assertSame('40001', $context['sql_state']);
$this->assertSame(1213, $context['driver_code']);
$this->assertStringNotContainsString('private', json_encode($context));
return true;
});
$this->assertSame('保留', DB::connection('business_mirror')->table('test_records')->value('displayName'));
}
public function test_project_dataset_registry_has_unique_safe_targets(): void
{
$configured = require base_path('config/upstream_mirrors.php');
@ -561,7 +658,7 @@ class BusinessMirrorSynchronizerTest extends TestCase
class FakeBusinessMirrorQuery extends UpstreamReadOnlyQuery
{
/** @var list<list<array<string,mixed>>> */
/** @var list<list<array<string,mixed>>|\Throwable> */
public array $batches = [];
/** @var list<array{sql:string,bindings:array}> */
@ -576,6 +673,11 @@ class FakeBusinessMirrorQuery extends UpstreamReadOnlyQuery
{
$this->calls[] = compact('sql', 'bindings');
return array_shift($this->batches) ?? [];
$batch = array_shift($this->batches) ?? [];
if ($batch instanceof \Throwable) {
throw $batch;
}
return $batch;
}
}