数据库变更管理
大约 15 分钟约 4509 字
数据库变更管理
简介
数据库变更管理（Database Change Management）是将数据库 Schema 变更、数据迁移和配置调整纳入版本控制和自动化流程的实践。与应用程序代码不同，数据库变更具有状态性——你不能简单地用新版本覆盖旧版本，而是必须将数据库从一个状态转换到另一个状态。
缺乏规范的数据库变更管理会导致诸多问题：生产环境与开发环境不一致、变更回滚困难、部署失败、数据丢失等。通过引入数据库迁移工具和标准化流程，团队可以实现数据库变更的可追溯、可重复和可回滚，从而降低部署风险，提高发布效率。
特点
- 版本控制：每次变更都有明确的版本标识和变更记录
- 可重复执行：在任何环境上都能重现相同的数据库状态
- 可回滚：支持变更的逆向操作
- 自动化：集成到 CI/CD 流水线中，减少人工操作
- 协作友好：多人并行开发时自动处理变更冲突
数据库迁移工具对比
Flyway
Flyway 是最流行的开源数据库迁移工具之一，以简单直接著称。
-- Flyway 迁移脚本命名规范
-- V{version}__{description}.sql
-- version 格式：点分版本号，如 1.0、1.1、2.0.3
-- V1.0__Create_Orders_table.sql
CREATE TABLE Orders (
OrderId INT PRIMARY KEY AUTO_INCREMENT,
CustomerId INT NOT NULL,
OrderDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
TotalAmount DECIMAL(18, 2) NOT NULL DEFAULT 0,
Status VARCHAR(20) NOT NULL DEFAULT 'New',
CreatedAt DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
UpdatedAt DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
INDEX IX_Orders_CustomerId (CustomerId),
INDEX IX_Orders_Status (Status),
INDEX IX_Orders_CreatedAt (CreatedAt)
);
-- V1.1__Create_OrderItems_table.sql
CREATE TABLE OrderItems (
OrderItemId INT PRIMARY KEY AUTO_INCREMENT,
OrderId INT NOT NULL,
ProductId INT NOT NULL,
Quantity INT NOT NULL,
UnitPrice DECIMAL(18, 2) NOT NULL,
TotalPrice DECIMAL(18, 2) GENERATED ALWAYS AS (Quantity * UnitPrice) STORED,
FOREIGN KEY (OrderId) REFERENCES Orders(OrderId),
INDEX IX_OrderItems_OrderId (OrderId),
INDEX IX_OrderItems_ProductId (ProductId)
);-- V1.2__Add_Order_shipping_info.sql
ALTER TABLE Orders ADD COLUMN ShippingAddress VARCHAR(500);
ALTER TABLE Orders ADD COLUMN TrackingNumber VARCHAR(100);
ALTER TABLE Orders ADD COLUMN ShippedAt DATETIME;
-- V2.0__Add_Customer_tier.sql
ALTER TABLE Customers ADD COLUMN Tier VARCHAR(20) NOT NULL DEFAULT 'Standard';
ALTER TABLE Customers ADD COLUMN TierUpdatedAt DATETIME;
-- 更新现有客户等级
UPDATE Customers c
SET Tier = CASE
WHEN (SELECT SUM(TotalAmount) FROM Orders WHERE CustomerId = c.CustomerId) > 10000 THEN 'Gold'
WHEN (SELECT SUM(TotalAmount) FROM Orders WHERE CustomerId = c.CustomerId) > 5000 THEN 'Silver'
ELSE 'Standard'
END,
TierUpdatedAt = CURRENT_TIMESTAMP;-- Flyway 配置文件 flyway.conf
flyway.url=jdbc:mysql://localhost:3306/mydb
flyway.user=root
flyway.password=your_password
flyway.locations=classpath:db/migration
flyway.baselineOnMigrate=true
flyway.baselineVersion=1.0
flyway.outOfOrder=false
flyway.validateOnMigrate=true
flyway.encoding=UTF-8
flyway.table=flyway_schema_history# Flyway 命令行操作
# 查看迁移状态
flyway info
# 执行迁移
flyway migrate
# 验证迁移完整性
flyway validate
# 清理数据库（危险！仅用于开发环境）
flyway clean
# 修复迁移历史表
flyway repairLiquibase
Liquibase 支持多种格式（XML、YAML、JSON）定义变更，功能更为丰富。
# liquibase changelog (db/changelog/db.changelog-master.yaml)
databaseChangeLog:
- include:
file: db/changelog/changes/001-create-orders.yaml
- include:
file: db/changelog/changes/002-create-order-items.yaml
- include:
file: db/changelog/changes/003-add-shipping-info.yaml# db/changelog/changes/001-create-orders.yaml
databaseChangeLog:
- changeSet:
id: 001-create-orders
author: team-lead
comment: 创建订单表
changes:
- createTable:
tableName: Orders
columns:
- column:
name: OrderId
type: INT
autoIncrement: true
constraints:
primaryKey: true
nullable: false
- column:
name: CustomerId
type: INT
constraints:
nullable: false
- column:
name: OrderDate
type: DATETIME
defaultValueComputed: CURRENT_TIMESTAMP
constraints:
nullable: false
- column:
name: TotalAmount
type: DECIMAL(18, 2)
defaultValueNumeric: 0
constraints:
nullable: false
- column:
name: Status
type: VARCHAR(20)
defaultValue: New
constraints:
nullable: false
- column:
name: CreatedAt
type: DATETIME
defaultValueComputed: CURRENT_TIMESTAMP
constraints:
nullable: false
- createIndex:
tableName: Orders
indexName: IX_Orders_CustomerId
columns:
- column:
name: CustomerId
- createIndex:
tableName: Orders
indexName: IX_Orders_Status
columns:
- column:
name: Status
rollback:
- dropTable:
tableName: Orders# 带前置条件和回滚的变更集
databaseChangeLog:
- changeSet:
id: 004-add-customer-tier
author: developer
preconditions:
- onFail: MARK_RAN
- tableExists:
tableName: Customers
changes:
- addColumn:
tableName: Customers
columns:
- column:
name: Tier
type: VARCHAR(20)
defaultValue: Standard
constraints:
nullable: false
- column:
name: TierUpdatedAt
type: DATETIME
- sql:
comment: 根据历史消费更新客户等级
sql: |
UPDATE Customers c
SET Tier = CASE
WHEN (SELECT COALESCE(SUM(TotalAmount), 0) FROM Orders WHERE CustomerId = c.CustomerId) > 10000 THEN 'Gold'
WHEN (SELECT COALESCE(SUM(TotalAmount), 0) FROM Orders WHERE CustomerId = c.CustomerId) > 5000 THEN 'Silver'
ELSE 'Standard'
END,
TierUpdatedAt = CURRENT_TIMESTAMP
rollback:
- dropColumn:
tableName: Customers
columns:
- column:
name: Tier
- column:
name: TierUpdatedAt# Liquibase 命令行操作
# 执行迁移
liquibase update
# 查看未执行的变更集
liquibase status
# 回滚到指定标签
liquibase rollback v1.0
# 回滚指定数量的变更集
liquibase rollbackCount 1
# 生成 SQL（不执行）
liquibase updateSQL > output.sql
# 生成差异报告
liquibase diff --format=txtEF Core Migrations
// EF Core 迁移是 .NET 生态的首选方案
public class AppDbContext : DbContext
{
public DbSet<Order> Orders { get; set; }
public DbSet<OrderItem> OrderItems { get; set; }
public DbSet<Customer> Customers { get; set; }
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
modelBuilder.Entity<Order>(entity =>
{
entity.HasKey(e => e.OrderId);
entity.Property(e => e.Status).HasMaxLength(20).HasDefaultValue("New");
entity.Property(e => e.TotalAmount).HasPrecision(18, 2);
entity.HasIndex(e => e.CustomerId);
entity.HasIndex(e => e.Status);
entity.HasIndex(e => e.CreatedAt);
});
modelBuilder.Entity<OrderItem>(entity =>
{
entity.HasKey(e => e.OrderItemId);
entity.Property(e => e.UnitPrice).HasPrecision(18, 2);
entity.HasOne(e => e.Order)
.WithMany(o => o.Items)
.HasForeignKey(e => e.OrderId);
entity.HasIndex(e => e.ProductId);
});
}
}# EF Core 迁移命令
# 创建新迁移
dotnet ef migrations add CreateOrdersTable
# 更新数据库到最新版本
dotnet ef database update
# 更新到指定迁移版本
dotnet ef database update CreateOrdersTable
# 回滚到指定版本
dotnet ef database update 20240101000000_InitialCreate
# 移除最近一次未应用的迁移
dotnet ef migrations remove
# 生成 SQL 脚本（不执行）
dotnet ef migrations script --output migrations.sql
dotnet ef migrations script AddNewTable --idempotent > idempotent_migration.sql
# 查看迁移列表
dotnet ef migrations list// 自动生成的迁移文件示例
public partial class AddCustomerTier : Migration
{
protected override void Up(MigrationBuilder migrationBuilder)
{
migrationBuilder.AddColumn<string>(
name: "Tier",
table: "Customers",
type: "nvarchar(20)",
maxLength: 20,
nullable: false,
defaultValue: "Standard");
migrationBuilder.AddColumn<DateTime>(
name: "TierUpdatedAt",
table: "Customers",
type: "datetime2",
nullable: true);
migrationBuilder.Sql(@"
UPDATE Customers c
SET Tier = CASE
WHEN (SELECT COALESCE(SUM(o.TotalAmount), 0) FROM Orders o WHERE o.CustomerId = c.CustomerId) > 10000 THEN 'Gold'
WHEN (SELECT COALESCE(SUM(o.TotalAmount), 0) FROM Orders o WHERE o.CustomerId = c.CustomerId) > 5000 THEN 'Silver'
ELSE 'Standard'
END,
TierUpdatedAt = GETUTCDATE()");
}
protected override void Down(MigrationBuilder migrationBuilder)
{
migrationBuilder.DropColumn(name: "TierUpdatedAt", table: "Customers");
migrationBuilder.DropColumn(name: "Tier", table: "Customers");
}
}迁移版本管理策略
版本号规范
版本号策略对比：
1. 顺序递增（Flyway 默认）
V1__initial.sql
V2__add_orders.sql
V3__add_shipping.sql
优点：简单直观
缺点：并行开发时版本号冲突
2. 时间戳（推荐）
V20240115120000__initial.sql
V20240116103000__add_orders.sql
V20240117091500__add_shipping.sql
优点：避免冲突，按时间排序
缺点：版本号较长
3. 语义版本 + 描述
V1.0.0__initial_schema.sql
V1.1.0__add_order_management.sql
V1.1.1__fix_order_status_index.sql
优点：语义清晰
缺点：需要团队约定
4. 分支感知版本（推荐团队协作）
V20240115_feature_orders__create_orders_table.sql
V20240116_feature_users__add_user_preferences.sql
V20240117_hotfix_billing__fix_amount_precision.sql
优点：可追溯变更来源
缺点：命名较长迁移目录结构
db/
├── migration/
│ ├── V1.0__initial/
│ │ ├── V1.0.1__create_users_table.sql
│ │ ├── V1.0.2__create_products_table.sql
│ │ └── V1.0.3__seed_data.sql
│ ├── V1.1__order_management/
│ │ ├── V1.1.1__create_orders_table.sql
│ │ └── V1.1.2__create_order_items_table.sql
│ └── V2.0__customer_tier/
│ ├── V2.0.1__add_tier_column.sql
│ └── V2.0.2__update_tier_data.sql
├── rollback/
│ ├── V1.1__order_management/
│ │ ├── U1.1.2__drop_order_items_table.sql
│ │ └── U1.1.1__drop_orders_table.sql
│ └── V2.0__customer_tier/
│ └── U2.0.1__remove_tier_column.sql
└── repeatable/
└── R__create_useful_views.sql回滚策略
回滚策略对比
回滚策略：
策略1：补偿式回滚（推荐）
├── 为每个迁移编写逆向脚本
├── 回滚时执行逆向操作
└── 适用：大多数场景
策略2：备份恢复
├── 变更前完整备份数据库
├── 回滚时恢复备份
└── 适用：大规模结构变更
策略3：蓝绿切换
├── 维护两套数据库
├── 切换流量指向旧版本
└── 适用：零停机要求高
策略4：前向修复
├── 不回滚，发布修复迁移
├── 解决问题并继续前进
└── 适用：已发生数据变更// 回滚管理器
public class MigrationRollbackManager
{
private readonly string _connectionString;
public MigrationRollbackManager(string connectionString)
{
_connectionString = connectionString;
}
public async Task RollbackToVersion(string targetVersion)
{
using var connection = new SqlConnection(_connectionString);
await connection.OpenAsync();
// 获取已执行的迁移
var appliedMigrations = await GetAppliedMigrationsAsync(connection);
// 找出需要回滚的迁移
var migrationsToRollback = appliedMigrations
.Where(m => string.Compare(m.Version, targetVersion) > 0)
.OrderByDescending(m => m.Version)
.ToList();
using var transaction = connection.BeginTransaction();
try
{
foreach (var migration in migrationsToRollback)
{
Console.WriteLine($"正在回滚迁移: {migration.Version} - {migration.Description}");
string rollbackSql = await LoadRollbackScriptAsync(migration.Version);
using var cmd = new SqlCommand(rollbackSql, connection, transaction);
await cmd.ExecuteNonQueryAsync();
// 从历史表删除记录
using var deleteCmd = new SqlCommand(
"DELETE FROM __MigrationHistory WHERE Version = @version",
connection, transaction);
deleteCmd.Parameters.AddWithValue("@version", migration.Version);
await deleteCmd.ExecuteNonQueryAsync();
}
await transaction.CommitAsync();
Console.WriteLine("回滚完成");
}
catch (Exception ex)
{
await transaction.RollbackAsync();
Console.WriteLine($"回滚失败: {ex.Message}");
throw;
}
}
private async Task<List<MigrationRecord>> GetAppliedMigrationsAsync(
SqlConnection connection)
{
var results = new List<MigrationRecord>();
using var cmd = new SqlCommand(
"SELECT Version, Description, AppliedAt FROM __MigrationHistory ORDER BY Version",
connection);
using var reader = await cmd.ExecuteReaderAsync();
while (await reader.ReadAsync())
{
results.Add(new MigrationRecord
{
Version = reader.GetString(0),
Description = reader.GetString(1),
AppliedAt = reader.GetDateTime(2)
});
}
return results;
}
private async Task<string> LoadRollbackScriptAsync(string version)
{
string path = $"db/rollback/U{version}__rollback.sql";
return await File.ReadAllTextAsync(path);
}
}
public class MigrationRecord
{
public string Version { get; set; }
public string Description { get; set; }
public DateTime AppliedAt { get; set; }
}零停机迁移
大表在线 DDL
-- MySQL 在线 DDL（Online DDL）
-- 添加列（不锁表）
ALTER TABLE Orders
ADD COLUMN ShippingMethod VARCHAR(50),
ALGORITHM=INPLACE, LOCK=NONE;
-- 添加索引（不锁表）
ALTER TABLE Orders
ADD INDEX IX_Orders_ShippingMethod (ShippingMethod),
ALGORITHM=INPLACE, LOCK=NONE;
-- MySQL 8.0 即时 DDL（Instant DDL）
-- 添加列在表末尾，几乎瞬间完成
ALTER TABLE Orders
ADD COLUMN Notes VARCHAR(500),
ALGORITHM=INSTANT;-- pt-online-schema-change（Percona Toolkit）
-- 在线修改大表结构
pt-online-schema-change \
--alter "ADD COLUMN ShippingMethod VARCHAR(50)" \
--host=localhost \
--user=root \
--password=your_password \
D=mydb,t=Orders \
--execute \
--max-load=Threads_running=25 \
--critical-load=Threads_running=50 \
--chunk-size=1000 \
--chunk-time=0.5
-- 原理：
-- 1. 创建与原表结构相同的新表（_Orders_new）
-- 2. 在新表上执行 ALTER
-- 3. 创建触发器同步增量数据
-- 4. 分批拷贝原表数据到新表
-- 5. 原子重命名表-- SQL Server 在线索引操作
CREATE INDEX IX_Orders_CustomerId_Status
ON Orders (CustomerId, Status)
WITH (ONLINE = ON, SORT_IN_TEMPDB = ON);
-- SQL Server 在线添加列（包含默认值）
-- SQL Server 2012+ 支持带默认值的在线添加列
ALTER TABLE Orders ADD ShippingMethod VARCHAR(50) NOT NULL
CONSTRAINT DF_Orders_ShippingMethod DEFAULT 'Standard'
WITH VALUES;蓝绿数据库部署
蓝绿数据库部署流程：
初始状态：
应用 (v1) → 蓝数据库 (v1 schema)
部署步骤：
1. 克隆蓝数据库到绿数据库
应用 (v1) → 蓝数据库 (v1 schema)
绿数据库 (v1 schema)
2. 在绿数据库执行迁移
应用 (v1) → 蓝数据库 (v1 schema)
绿数据库 (v2 schema)
3. 部署新版本应用到绿环境
应用 (v1) → 蓝数据库 (v1 schema)
应用 (v2) → 绿数据库 (v2 schema)
4. 切换流量
应用 (v2) → 绿数据库 (v2 schema)
蓝数据库保留作为回滚备份
5. 确认稳定后清理
应用 (v2) → 绿数据库 (v2 schema) [重命名为蓝]向后兼容的变更模式
-- 向后兼容的变更模式（Expand-Contract 模式）
-- 阶段1：扩展（Expand）- 添加新列/表，不删除旧的
-- 添加新的名字列
ALTER TABLE Customers ADD COLUMN FullName NVARCHAR(200);
-- 部署双写代码（同时写入 FirstName/LastName 和 FullName）
-- 等待所有实例都更新
-- 回填数据
UPDATE Customers SET FullName = CONCAT(FirstName, ' ', LastName)
WHERE FullName IS NULL;
-- 阶段2：迁移（Migrate）- 切换读取源
-- 应用代码改为从 FullName 读取
-- 阶段3：收缩（Contract）- 删除旧列
ALTER TABLE Customers DROP COLUMN FirstName;
ALTER TABLE Customers DROP COLUMN LastName;// .NET 中实现向后兼容的迁移策略
public class BackwardCompatibleMigration
{
// 阶段1：扩展 - 添加新字段
public void Expand(MigrationBuilder migrationBuilder)
{
migrationBuilder.AddColumn<string>(
name: "FullName",
table: "Customers",
type: "nvarchar(200)",
maxLength: 200,
nullable: true); // 允许 NULL，因为旧代码不会写此字段
}
// 在应用中双写
public class CustomerService
{
public async Task CreateCustomer(Customer customer)
{
// 同时写入新旧字段
var entity = new CustomerEntity
{
FirstName = customer.FirstName,
LastName = customer.LastName,
FullName = $"{customer.FirstName} {customer.LastName}"
};
_context.Customers.Add(entity);
await _context.SaveChangesAsync();
}
}
// 阶段3：收缩 - 删除旧字段
public void Contract(MigrationBuilder migrationBuilder)
{
migrationBuilder.DropColumn(name: "FirstName", table: "Customers");
migrationBuilder.DropColumn(name: "LastName", table: "Customers");
}
}数据迁移验证
// 数据迁移验证框架
public class MigrationValidator
{
private readonly string _sourceConnectionString;
private readonly string _targetConnectionString;
public MigrationValidator(
string sourceConnectionString,
string targetConnectionString)
{
_sourceConnectionString = sourceConnectionString;
_targetConnectionString = targetConnectionString;
}
public async Task<ValidationResult> Validate(
MigrationValidationConfig config)
{
var result = new ValidationResult();
// 1. 数据量验证
await ValidateRowCounts(result);
// 2. 数据完整性验证
await ValidateDataIntegrity(result, config);
// 3. Schema 验证
await ValidateSchema(result);
// 4. 外键约束验证
await ValidateForeignKeys(result);
// 5. 索引验证
await ValidateIndexes(result);
return result;
}
private async Task ValidateRowCounts(ValidationResult result)
{
using var sourceConn = new SqlConnection(_sourceConnectionString);
using var targetConn = new SqlConnection(_targetConnectionString);
await sourceConn.OpenAsync();
await targetConn.OpenAsync();
var tables = new[] { "Orders", "OrderItems", "Customers", "Products" };
foreach (var table in tables)
{
var sourceCount = await GetRowCount(sourceConn, table);
var targetCount = await GetRowCount(targetConn, table);
if (sourceCount != targetCount)
{
result.Errors.Add(
$"表 {table} 行数不匹配：源 {sourceCount}，目标 {targetCount}");
}
else
{
result.Successes.Add($"表 {table} 行数验证通过：{sourceCount} 行");
}
}
}
private async Task ValidateDataIntegrity(
ValidationResult result,
MigrationValidationConfig config)
{
using var sourceConn = new SqlConnection(_sourceConnectionString);
using var targetConn = new SqlConnection(_targetConnectionString);
await sourceConn.OpenAsync();
await targetConn.OpenAsync();
// 抽样验证数据一致性
foreach (var table in config.TablesToValidate)
{
string sampleQuery = $@"
SELECT TOP {config.SampleSize} *
FROM {table}
ORDER BY NEWID()";
var sourceData = await QueryToDictionary(sourceConn, sampleQuery, table);
var targetData = await QueryToDictionary(targetConn, sampleQuery, table);
foreach (var kvp in sourceData)
{
if (!targetData.TryGetValue(kvp.Key, out var targetRow))
{
result.Errors.Add($"{table} 缺少记录 ID={kvp.Key}");
continue;
}
foreach (var field in kvp.Value)
{
if (!Equals(field.Value, targetRow[field.Key]))
{
result.Errors.Add(
$"{table} ID={kvp.Key} 字段 {field.Key} 不匹配：" +
$"源={field.Value}，目标={targetRow[field.Key]}");
}
}
}
}
}
private async Task<long> GetRowCount(SqlConnection connection, string table)
{
using var cmd = new SqlCommand(
$"SELECT COUNT_BIG(*) FROM {table}", connection);
return (long)await cmd.ExecuteScalarAsync();
}
private async Task<Dictionary<string, Dictionary<string, object>>>
QueryToDictionary(SqlConnection connection, string query, string table)
{
var result = new Dictionary<string, Dictionary<string, object>>();
using var cmd = new SqlCommand(query, connection);
using var reader = await cmd.ExecuteReaderAsync();
while (await reader.ReadAsync())
{
var row = new Dictionary<string, object>();
for (int i = 0; i < reader.FieldCount; i++)
{
row[reader.GetName(i)] = reader.GetValue(i);
}
string key = row.ContainsKey("Id") ? row["Id"].ToString() :
row.FirstOrDefault().Value?.ToString() ?? Guid.NewGuid().ToString();
result[key] = row;
}
return result;
}
}
public class MigrationValidationConfig
{
public string[] TablesToValidate { get; set; }
public int SampleSize { get; set; } = 1000;
}
public class ValidationResult
{
public List<string> Errors { get; set; } = new();
public List<string> Successes { get; set; } = new();
public bool IsValid => !Errors.Any();
}CI/CD 集成
GitHub Actions 集成
# .github/workflows/database-migration.yml
name: Database Migration
on:
push:
branches: [main]
paths:
- 'db/migrations/**'
- 'src/**/Migrations/**'
jobs:
validate-migration:
runs-on: ubuntu-latest
services:
mysql:
image: mysql:8.0
env:
MYSQL_ROOT_PASSWORD: test
MYSQL_DATABASE: testdb
ports:
- 3306:3306
steps:
- uses: actions/checkout@v4
- name: Validate SQL syntax
run: |
for file in db/migrations/*.sql; do
echo "Validating $file"
mysql -h 127.0.0.1 -u root -ptest testdb < "$file" || exit 1
echo "ROLLBACK;" | mysql -h 127.0.0.1 -u root -ptest testdb
done
- name: Check migration naming
run: |
python3 scripts/validate_migration_naming.py db/migrations/
- name: Generate migration script
run: |
flyway migrate -url=jdbc:mysql://127.0.0.1:3306/testdb \
-user=root -password=test \
-locations=filesystem:db/migrations \
-validateOnMigrate=true
deploy-migration:
needs: validate-migration
runs-on: ubuntu-latest
environment: production
steps:
- uses: actions/checkout@v4
- name: Backup database
run: |
mysqldump -h ${{ secrets.DB_HOST }} -u ${{ secrets.DB_USER }} \
-p${{ secrets.DB_PASSWORD }} ${{ secrets.DB_NAME }} \
> backup_$(date +%Y%m%d_%H%M%S).sql
- name: Run migration with rollback on failure
run: |
if ! flyway migrate \
-url=jdbc:mysql://${{ secrets.DB_HOST }}:3306/${{ secrets.DB_NAME }} \
-user=${{ secrets.DB_USER }} \
-password=${{ secrets.DB_PASSWORD }}; then
echo "Migration failed, rolling back..."
flyway undo \
-url=jdbc:mysql://${{ secrets.DB_HOST }}:3306/${{ secrets.DB_NAME }} \
-user=${{ secrets.DB_USER }} \
-password=${{ secrets.DB_PASSWORD }}
exit 1
fi
- name: Validate migration
run: |
flyway validate \
-url=jdbc:mysql://${{ secrets.DB_HOST }}:3306/${{ secrets.DB_NAME }} \
-user=${{ secrets.DB_USER }} \
-password=${{ secrets.DB_PASSWORD }}Azure DevOps Pipeline 集成
# azure-pipelines-db-migration.yml
trigger:
paths:
include:
- 'src/**/Migrations/*'
variables:
- group: database-credentials
stages:
- stage: Test
jobs:
- job: MigrationTest
pool:
vmImage: 'ubuntu-latest'
steps:
- task: UseDotNet@2
inputs:
packageType: 'sdk'
version: '8.0.x'
- script: |
dotnet tool install --global dotnet-ef
dotnet ef migrations script --project src/App --idempotent \
--output $(Build.ArtifactStagingDirectory)/migration.sql
displayName: 'Generate migration SQL'
- script: |
dotnet ef database update --project src/App \
--connection "Server=localhost;Database=testdb;User=sa;Password=Test123!"
displayName: 'Test migration on clean database'
- stage: Deploy
dependsOn: Test
condition: succeeded()
jobs:
- deployment: DeployMigration
environment: production
strategy:
runOnce:
deploy:
steps:
- script: |
dotnet ef database update --project src/App \
--connection "$(ConnectionString)"
displayName: 'Apply migration to production'
env:
ConnectionString: $(ProductionConnectionString)迁移测试
// 数据库迁移集成测试
public class MigrationTests
{
[Fact]
public async Task AllMigrations_ShouldApplySuccessfully()
{
// 使用独立的测试数据库
var connectionString = "Server=localhost;Database=MigrationTestDb;" +
"User=sa;Password=Test123!;TrustServerCertificate=True";
// 确保从空数据库开始
await ResetDatabase(connectionString);
using var context = new AppDbContext(
new DbContextOptionsBuilder<AppDbContext>()
.UseSqlServer(connectionString)
.Options);
// 执行所有迁移
var pendingMigrations = await context.Database.GetPendingMigrationsAsync();
await context.Database.MigrateAsync();
// 验证所有迁移都已应用
var appliedMigrations = await context.Database.GetAppliedMigrationsAsync();
Assert.Equal(pendingMigrations.Count(), appliedMigrations.Count());
}
[Fact]
public async Task Migration_ShouldBeIdempotent()
{
var connectionString = "Server=localhost;Database=IdempotentTestDb;" +
"User=sa;Password=Test123!;TrustServerCertificate=True";
await ResetDatabase(connectionString);
using var context = new AppDbContext(
new DbContextOptionsBuilder<AppDbContext>()
.UseSqlServer(connectionString)
.Options);
// 第一次执行
await context.Database.MigrateAsync();
// 第二次执行（不应抛出异常）
await context.Database.MigrateAsync();
// 验证数据完整性
Assert.True(await context.Database.CanConnectAsync());
}
[Fact]
public async Task SeedData_ShouldBePresent_AfterMigration()
{
var connectionString = "Server=localhost;Database=SeedTestDb;" +
"User=sa;Password=Test123!;TrustServerCertificate=True";
await ResetDatabase(connectionString);
using var context = new AppDbContext(
new DbContextOptionsBuilder<AppDbContext>()
.UseSqlServer(connectionString)
.Options);
await context.Database.MigrateAsync();
// 验证种子数据
var roles = await context.Roles.ToListAsync();
Assert.Contains(roles, r => r.Name == "Admin");
Assert.Contains(roles, r => r.Name == "User");
}
private async Task ResetDatabase(string connectionString)
{
using var connection = new SqlConnection(
connectionString.Replace("MigrationTestDb", "master")
.Replace("IdempotentTestDb", "master")
.Replace("SeedTestDb", "master"));
await connection.OpenAsync();
var dbNames = new[] { "MigrationTestDb", "IdempotentTestDb", "SeedTestDb" };
foreach (var dbName in dbNames)
{
using var cmd = new SqlCommand(
$"IF DB_ID('{dbName}') IS NOT NULL DROP DATABASE [{dbName}]", connection);
await cmd.ExecuteNonQueryAsync();
}
}
}总结
数据库变更管理是现代软件交付中不可或缺的环节。通过规范化的迁移工具和流程，团队可以实现：
- 可追溯：每次变更都有记录，可以回溯历史
- 可重复：在任何环境上都能重建相同的数据库状态
- 可回滚：变更失败时有明确的恢复路径
- 自动化：减少人工操作，降低出错概率
- 协作友好：多人并行开发不会互相冲突
关键知识点
- Flyway 和 Liquibase 是主流的数据库迁移工具
- EF Core Migrations 是 .NET 项目的首选方案
- 迁移脚本应纳入版本控制
- 大表变更需要使用在线 DDL 工具
- Expand-Contract 模式可以实现零停机迁移
- 每次迁移都应有对应的回滚脚本
常见误区
误区1：迁移脚本不需要测试
迁移脚本同样是代码，需要在测试环境验证。一个语法错误的迁移脚本会导致整个部署失败。
误区2：生产环境可以直接手动执行 SQL
手动操作无法追溯、无法重复、无法自动回滚。所有变更都应通过迁移工具执行。
误区3：ALTER TABLE 不会影响生产
在大表上执行 ALTER TABLE 可能导致长时间锁表，需要使用在线 DDL 或 pt-online-schema-change。
误区4：迁移回滚总是可行的
数据变更（如删除数据）通常无法回滚。设计迁移时需要考虑回滚的可行性。
误区5：忽略迁移的向后兼容性
部署过程中新旧版本代码会同时运行，迁移必须与新旧版本代码都兼容。
进阶路线
- 状态迁移 vs 变更迁移：理解 declarative migration 的概念
- 数据库 CI/CD 成熟度模型：从手动到全自动的演进路径
- 多数据库迁移：同一应用支持多种数据库的迁移策略
- 大规模数据迁移：ETL 工具与迁移工具的配合
- 数据库即代码（Database as Code）：将数据库完全纳入版本控制
适用场景
- 持续交付：每次部署自动执行数据库迁移
- 多环境管理：开发、测试、预发布、生产环境保持一致
- 团队协作：多人并行开发不同的数据库变更
- 合规审计：追踪所有数据库变更记录
- 灾难恢复：从备份恢复后快速将数据库更新到最新版本
落地建议
- 选择适合团队的迁移工具（.NET 项目推荐 EF Core Migrations）
- 建立迁移脚本命名规范和目录结构
- 每个迁移都必须有对应的回滚脚本
- 在 CI/CD 中自动验证迁移脚本
- 大表变更提前评估执行时间和锁影响
- 使用 Expand-Contract 模式处理破坏性变更
- 生产环境部署前在预发布环境完整演练
排错清单
复盘问题
- 你的团队使用什么工具管理数据库变更？
- 上次数据库部署失败是什么原因？如何改进的？
- 你是否有定期清理旧迁移的策略？
- 大表变更的执行计划是否提前评估？
- 你的迁移流程是否支持蓝绿部署？
延伸阅读
- Flyway Documentation
- Liquibase Documentation
- EF Core Migrations
- Evolutionary Database Design
- 《Database Reliability Engineering》- Laine Campbell & Charity Majors
- 《Refactoring Databases》- Scott Ambler & Pramod Sadalage