企业官网建设流程全解析

一、数据一致性概述

二、迁移过程中的一致性风险

2.1 一致性风险类型

2.2 风险影响评估

三、一致性保障策略

3.1 事务保障

事务迁移流程：

事务隔离级别：

3.2 幂等性保障

幂等性设计：

幂等性代码示例：

import uuid class IdempotentMigration: def __init__(self, db): self.db = db def generate_request_id(self): return str(uuid.uuid4()) def check_request_id(self, request_id): result = self.db.execute( "SELECT COUNT(*) FROM migration_requests WHERE request_id = %s", (request_id,) ) return result.fetchone()[0] > 0 def save_request_id(self, request_id): self.db.execute( "INSERT INTO migration_requests (request_id, status, created_at) VALUES (%s, %s, NOW())", (request_id, 'processing') ) def update_request_status(self, request_id, status): self.db.execute( "UPDATE migration_requests SET status = %s, updated_at = NOW() WHERE request_id = %s", (status, request_id) ) def migrate_with_idempotency(self, data, request_id=None): if request_id is None: request_id = self.generate_request_id() if self.check_request_id(request_id): print(f"Request {request_id} already processed") return {'success': True, 'message': 'Already processed'} try: self.save_request_id(request_id) with self.db.begin(): self._migrate_data(data) self.update_request_status(request_id, 'success') return {'success': True, 'message': 'Migration successful'} except Exception as e: self.update_request_status(request_id, 'failed') return {'success': False, 'message': str(e)} def _migrate_data(self, data): for record in data: existing = self.db.execute( "SELECT id FROM target_table WHERE id = %s", (record['id'],) ).fetchone() if existing: self.db.execute( "UPDATE target_table SET name = %s, email = %s WHERE id = %s", (record['name'], record['email'], record['id']) ) else: self.db.execute( "INSERT INTO target_table (id, name, email) VALUES (%s, %s, %s)", (record['id'], record['name'], record['email']) )

3.3 分布式锁

分布式锁实现：

Redis分布式锁代码示例：

import redis import time class DistributedLock: def __init__(self, redis_client, lock_key, expire_time=30): self.redis = redis_client self.lock_key = lock_key self.expire_time = expire_time self.lock_value = None def acquire(self, timeout=10): end_time = time.time() + timeout while time.time() < end_time: self.lock_value = str(uuid.uuid4()) result = self.redis.set( self.lock_key, self.lock_value, nx=True, ex=self.expire_time ) if result: return True time.sleep(0.1) return False def release(self): if self.lock_value: script = """ if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end """ self.redis.eval(script, 1, self.lock_key, self.lock_value) def __enter__(self): self.acquire() return self def __exit__(self, exc_type, exc_val, exc_tb): self.release()

3.4 数据校验点

校验点机制：

校验点实现：

class CheckpointMigration: def __init__(self, db): self.db = db def get_checkpoint(self, migration_id): result = self.db.execute( "SELECT position, status FROM migration_checkpoints WHERE migration_id = %s", (migration_id,) ).fetchone() if result: return {'position': result[0], 'status': result[1]} return {'position': 0, 'status': 'pending'} def set_checkpoint(self, migration_id, position, status): existing = self.db.execute( "SELECT id FROM migration_checkpoints WHERE migration_id = %s", (migration_id,) ).fetchone() if existing: self.db.execute( "UPDATE migration_checkpoints SET position = %s, status = %s, updated_at = NOW() WHERE migration_id = %s", (position, status, migration_id) ) else: self.db.execute( "INSERT INTO migration_checkpoints (migration_id, position, status, created_at) VALUES (%s, %s, %s, NOW())", (migration_id, position, status) ) def run_migration(self, migration_id, data, batch_size=1000): checkpoint = self.get_checkpoint(migration_id) start_pos = checkpoint['position'] total_records = len(data) print(f"Resuming migration from position {start_pos}") for i in range(start_pos, total_records, batch_size): batch = data[i:i+batch_size] try: with self.db.begin(): self._process_batch(batch) self.set_checkpoint(migration_id, i + len(batch), 'processing') print(f"Migrated {i + len(batch)}/{total_records} records") except Exception as e: print(f"Error at position {i}: {e}") self.set_checkpoint(migration_id, i, 'failed') raise self.set_checkpoint(migration_id, total_records, 'completed') print("Migration completed successfully") def _process_batch(self, batch): for record in batch: self.db.execute( "INSERT INTO target_table (id, name, email) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE name = %s, email = %s", (record['id'], record['name'], record['email'], record['name'], record['email']) )

四、数据校验机制

4.1 校验类型

4.2 数量校验

数量校验流程：

数量校验代码示例：

class CountValidator: def __init__(self, source_db, target_db): self.source_db = source_db self.target_db = target_db def get_source_count(self, table_name): result = self.source_db.execute(f"SELECT COUNT(*) FROM {table_name}") return result.fetchone()[0] def get_target_count(self, table_name): result = self.target_db.execute(f"SELECT COUNT(*) FROM {table_name}") return result.fetchone()[0] def validate(self, table_name): source_count = self.get_source_count(table_name) target_count = self.get_target_count(table_name) is_valid = source_count == target_count return { 'table': table_name, 'source_count': source_count, 'target_count': target_count, 'is_valid': is_valid, 'difference': source_count - target_count }

4.3 内容校验

内容校验流程：

内容校验代码示例：

import hashlib class ContentValidator: def __init__(self, source_db, target_db): self.source_db = source_db self.target_db = target_db def generate_record_hash(self, record): sorted_keys = sorted(record.keys()) hash_string = '|'.join(f"{k}={record[k]}" for k in sorted_keys) return hashlib.md5(hash_string.encode()).hexdigest() def get_sample_data(self, db, table_name, sample_size=100): result = db.execute( f"SELECT * FROM {table_name} ORDER BY id LIMIT %s", (sample_size,) ) columns = [desc[0] for desc in result.description] return [dict(zip(columns, row)) for row in result.fetchall()] def validate_sample(self, table_name, sample_size=100): source_sample = self.get_sample_data(self.source_db, table_name, sample_size) target_sample = self.get_sample_data(self.target_db, table_name, sample_size) mismatches = [] for source_record, target_record in zip(source_sample, target_sample): source_hash = self.generate_record_hash(source_record) target_hash = self.generate_record_hash(target_record) if source_hash != target_hash: mismatches.append({ 'id': source_record.get('id'), 'source': source_record, 'target': target_record }) return { 'table': table_name, 'sample_size': sample_size, 'mismatches': mismatches, 'is_valid': len(mismatches) == 0 } def validate_full(self, table_name): source_count = self.source_db.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0] target_count = self.target_db.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0] if source_count != target_count: return { 'table': table_name, 'is_valid': False, 'message': f"Count mismatch: source={source_count}, target={target_count}" } batch_size = 1000 mismatches = [] for offset in range(0, source_count, batch_size): source_batch = self.source_db.execute( f"SELECT * FROM {table_name} ORDER BY id LIMIT %s OFFSET %s", (batch_size, offset) ).fetchall() target_batch = self.target_db.execute( f"SELECT * FROM {table_name} ORDER BY id LIMIT %s OFFSET %s", (batch_size, offset) ).fetchall() source_columns = [desc[0] for desc in source_batch.description] target_columns = [desc[0] for desc in target_batch.description] for source_row, target_row in zip(source_batch.fetchall(), target_batch.fetchall()): source_record = dict(zip(source_columns, source_row)) target_record = dict(zip(target_columns, target_row)) source_hash = self.generate_record_hash(source_record) target_hash = self.generate_record_hash(target_record) if source_hash != target_hash: mismatches.append({ 'id': source_record.get('id'), 'source': source_record, 'target': target_record }) return { 'table': table_name, 'total_records': source_count, 'mismatches': mismatches, 'is_valid': len(mismatches) == 0 }

4.4 结构校验

结构校验代码示例：

class StructureValidator: def __init__(self, source_db, target_db): self.source_db = source_db self.target_db = target_db def get_table_structure(self, db, table_name): result = db.execute(f"DESCRIBE {table_name}") columns = [] for row in result.fetchall(): columns.append({ 'name': row[0], 'type': row[1], 'null': row[2], 'key': row[3], 'default': row[4], 'extra': row[5] }) return columns def validate(self, table_name): source_structure = self.get_table_structure(self.source_db, table_name) target_structure = self.get_table_structure(self.target_db, table_name) source_columns = {col['name']: col for col in source_structure} target_columns = {col['name']: col for col in target_structure} missing_columns = [name for name in source_columns if name not in target_columns] extra_columns = [name for name in target_columns if name not in source_columns] type_mismatches = [] for name in source_columns: if name in target_columns: source_type = source_columns[name]['type'] target_type = target_columns[name]['type'] if source_type != target_type: type_mismatches.append({ 'column': name, 'source_type': source_type, 'target_type': target_type }) return { 'table': table_name, 'missing_columns': missing_columns, 'extra_columns': extra_columns, 'type_mismatches': type_mismatches, 'is_valid': not missing_columns and not extra_columns and not type_mismatches }

五、一致性验证流程

5.1 验证流程

5.2 验证报告

验证报告示例：

{ "migration_id": "migration_001", "validation_time": "2024-01-15T10:00:00Z", "tables": [ { "name": "users", "count_validation": { "source_count": 10000, "target_count": 10000, "is_valid": true }, "content_validation": { "sample_size": 100, "mismatches": 0, "is_valid": true }, "structure_validation": { "missing_columns": [], "extra_columns": [], "type_mismatches": [], "is_valid": true }, "overall_status": "PASS" } ], "overall_status": "PASS", "total_tables": 1, "passed_tables": 1, "failed_tables": 0 }

六、常见问题

6.1 数据丢失

现象：目标数据少于源数据

解决方案：

6.2 数据重复

现象：目标数据存在重复记录

解决方案：

6.3 数据不一致

现象：源数据和目标数据不一致

解决方案：

6.4 完整性约束违反

现象：外键约束或唯一约束违反

解决方案：