并发控制 ConcurrencyCheck vs Timestamp

目录

• 并发控制基础
• ConcurrencyCheck 方式
• Timestamp 方式
• 两种方式对比
• .NET 8/9/10 新特性
• 最佳实践与陷阱

---

并发控制基础

什么是并发冲突?

并发冲突(Concurrency Conflict) 发生在多个用户同时修改同一条记录时,后提交的更改覆盖了先提交的更改,导致数据丢失。

时间线示例:

时刻 T1: 用户 A 读取产品(价格: $100)
时刻 T2: 用户 B 读取产品(价格: $100)
时刻 T3: 用户 A 修改价格为 $120,保存成功
时刻 T4: 用户 B 修改价格为 $110,保存成功

结果: 用户 A 的修改($120)被用户 B 的修改($110)覆盖!
用户 A 以为价格是 $120,实际是 $110 💥 数据不一致!

解决方案: 乐观并发控制

EF Core 使用乐观并发控制(Optimistic Concurrency Control),假设冲突很少发生,在保存时检测冲突。

// 核心原理: 检查数据是否被其他人修改
// 如果已修改,抛出 DbUpdateConcurrencyException

try
{
    await context.SaveChangesAsync();
}
catch (DbUpdateConcurrencyException ex)
{
    // 处理冲突: 
    // 1. 刷新数据,让用户重新决定
    // 2. 客户端获胜(强制覆盖)
    // 3. 数据库获胜(放弃修改)
    // 4. 合并变更
}

两种实现方式

方式	适用场景	优点	缺点
ConcurrencyCheck	监控特定字段	灵活,可选择字段	需要手动配置每个字段
Timestamp/RowVersion	监控整行	简单,自动管理	只能有一个时间戳列

---

ConcurrencyCheck 方式

基本概念

ConcurrencyCheck 特性标记在需要监控的属性上,EF Core 会在 UPDATE/DELETE 时将原始值包含在 WHERE 子句中。

public class Product
{
    public int Id { get; set; }
    public string Name { get; set; }
    
    [ConcurrencyCheck]  // ← 监控此字段
    public decimal Price { get; set; }
    
    public string Description { get; set; }  // 不监控
}

工作原理

-- 没有 ConcurrencyCheck
UPDATE Products SET Price = @p0 WHERE Id = @id;

-- 有 ConcurrencyCheck
UPDATE Products SET Price = @p0 
WHERE Id = @id AND Price = @original_price;  -- ← 检查原始值

-- 如果价格已被其他人修改,WHERE 条件不匹配,影响行数 = 0
-- EF Core 检测到影响行数为 0,抛出 DbUpdateConcurrencyException

配置方法

方法 1: 数据注解

using System.ComponentModel.DataAnnotations;

public class Product
{
    public int Id { get; set; }
    
    [ConcurrencyCheck]
    public decimal Price { get; set; }
    
    [ConcurrencyCheck]
    public int Stock { get; set; }
}

方法 2: Fluent API

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.Entity<Product>()
        .Property(p => p.Price)
        .IsConcurrencyToken();  // ← 等效于 [ConcurrencyCheck]
    
    modelBuilder.Entity<Product>()
        .Property(p => p.Stock)
        .IsConcurrencyToken();
}

完整示例

public class ProductService
{
    private readonly AppDbContext _context;
    
    public async Task UpdatePriceAsync(int productId, decimal newPrice)
    {
        var product = await _context.Products.FindAsync(productId);
        
        if (product == null)
            throw new NotFoundException($"Product {productId} not found");
        
        // 用户可能需要时间来决策,此时其他用户可能也在修改
        product.Price = newPrice;
        
        try
        {
            await _context.SaveChangesAsync();
            Console.WriteLine("价格更新成功");
        }
        catch (DbUpdateConcurrencyException ex)
        {
            // 获取数据库中的当前值
            var databaseEntry = ex.Entries.First().GetDatabaseValues();
            
            if (databaseEntry == null)
            {
                throw new InvalidOperationException("记录已被删除");
            }
            
            var currentPrice = databaseEntry.GetValue<decimal>("Price");
            
            Console.WriteLine($"并发冲突! 当前价格: ${currentPrice}");
            Console.WriteLine("请刷新页面,查看最新价格后重新编辑");
            
            // 刷新实体的值为数据库中的最新值
            ex.Entries.First().Reload(ReloadMode.StoreWins);
            
            throw;  // 或返回错误信息给前端
        }
    }
}

多字段监控

public class Order
{
    public int Id { get; set; }
    
    [ConcurrencyCheck]
    public decimal TotalAmount { get; set; }
    
    [ConcurrencyCheck]
    public string Status { get; set; }
    
    public DateTime OrderDate { get; set; }  // 不监控
}

// UPDATE Orders 
// SET TotalAmount = @p0, Status = @p1
// WHERE Id = @id 
//   AND TotalAmount = @original_total  -- 检查
//   AND Status = @original_status;     -- 检查

---

Timestamp 方式

基本概念

Timestamp/RowVersion 是数据库自动管理的二进制列,每次行数据变更时自动更新。这是推荐的方式。

public class Product
{
    public int Id { get; set; }
    public string Name { get; set; }
    public decimal Price { get; set; }
    
    [Timestamp]  // ← 或使用 byte[] RowVersion { get; set; }
    public byte[] RowVersion { get; set; }
}

工作原理

-- SQL Server: rowversion 类型(以前叫 timestamp)
CREATE TABLE Products (
    Id INT PRIMARY KEY IDENTITY(1,1),
    Name NVARCHAR(100),
    Price DECIMAL(18,2),
    RowVersion ROWVERSION NOT NULL  -- 自动管理
);

-- 插入时自动生成
INSERT INTO Products (Name, Price) VALUES ('Laptop', 999.99);
-- RowVersion 自动生成,如: 0x00000000000007D1

-- 更新时自动修改
UPDATE Products SET Price = 1099.99 WHERE Id = 1;
-- RowVersion 自动变为: 0x00000000000007D2

-- 并发检查
UPDATE Products SET Price = @p0 
WHERE Id = @id AND RowVersion = @original_version;

-- 如果 RowVersion 不匹配,影响行数 = 0,抛出异常

配置方法

方法 1: 数据注解

using System.ComponentModel.DataAnnotations;

public class Product
{
    public int Id { get; set; }
    public string Name { get; set; }
    
    [Timestamp]
    public byte[] RowVersion { get; set; }
}

方法 2: Fluent API

modelBuilder.Entity<Product>()
    .Property(p => p.RowVersion)
    .IsRowVersion();  // ← 标记为行版本

方法 3: 约定(推荐)

// EF Core 自动识别名为 "RowVersion" 或 "Timestamp" 的 byte[] 属性
public class Product
{
    public int Id { get; set; }
    public byte[] RowVersion { get; set; }  // 自动识别为时间戳
}

// 无需额外配置!

完整示例

public class ProductService
{
    private readonly AppDbContext _context;
    
    public async Task UpdateProductAsync(ProductDto dto, byte[] originalRowVersion)
    {
        var product = await _context.Products.FindAsync(dto.Id);
        
        if (product == null)
            throw new NotFoundException($"Product {dto.Id} not found");
        
        // 设置原始 RowVersion
        _context.Entry(product).OriginalValues["RowVersion"] = originalRowVersion;
        
        // 应用修改
        product.Name = dto.Name;
        product.Price = dto.Price;
        
        try
        {
            await _context.SaveChangesAsync();
            Console.WriteLine("更新成功");
        }
        catch (DbUpdateConcurrencyException)
        {
            Console.WriteLine("并发冲突! 数据已被其他人修改");
            throw;
        }
    }
}

// Minimal API 示例
app.MapPut("/api/products/{id}", async (
    int id, 
    ProductDto dto, 
    [FromHeader(Name = "If-Match")] string ifMatch,  // 从请求头获取 RowVersion
    AppDbContext db) =>
{
    var product = await db.Products.FindAsync(id);
    if (product == null)
        return Results.NotFound();
    
    // 解析 RowVersion
    var originalRowVersion = Convert.FromBase64String(ifMatch);
    db.Entry(product).OriginalValues["RowVersion"] = originalRowVersion;
    
    product.Name = dto.Name;
    product.Price = dto.Price;
    
    try
    {
        await db.SaveChangesAsync();
        return Results.NoContent();
    }
    catch (DbUpdateConcurrencyException)
    {
        return Results.Conflict("数据已被修改,请刷新后重试");
    }
});

前端集成

// TypeScript 客户端
interface Product {
  id: number;
  name: string;
  price: number;
  rowVersion: string;  // Base64 编码
}

async function updateProduct(product: Product) {
  const response = await fetch(`/api/products/${product.id}`, {
    method: 'PUT',
    headers: {
      'Content-Type': 'application/json',
      'If-Match': product.rowVersion  // 发送 RowVersion
    },
    body: JSON.stringify({
      name: product.name,
      price: product.price
    })
  });
  
  if (response.status === 409) {
    alert('数据已被其他人修改,请刷新页面');
    return null;
  }
  
  return await response.json();
}

---

两种方式对比

详细对比表

特性	ConcurrencyCheck	Timestamp/RowVersion
配置复杂度	中等(需指定字段)	简单(自动管理)
监控范围	指定字段	整行
数据库支持	所有数据库	SQL Server/Azure SQL 最佳
性能	好(WHERE 条件少)	最好(单列比较)
灵活性	高(可选择字段)	低(固定一列)
可靠性	中(可能漏掉字段)	高(自动监控所有变更)
推荐程度	⭐⭐⭐	⭐⭐⭐⭐⭐

选择决策树

graph TD
    A[需要并发控制?] --> B{数据库类型?}
    B -->|SQL Server/Azure| C[使用 Timestamp/RowVersion]
    B -->|PostgreSQL/MySQL| D{需要监控哪些字段?}
    D -->|所有字段| E[使用 Trigger + Timestamp]
    D -->|特定字段| F[使用 ConcurrencyCheck]
    C --> G[最佳实践]
    E --> G
    F --> G

PostgreSQL 和 MySQL 的支持

PostgreSQL

// PostgreSQL 不支持 rowversion,使用 xmin 系统列
public class Product
{
    public int Id { get; set; }
    public string Name { get; set; }
    
    [Timestamp]
    public uint Xmin { get; set; }  // PostgreSQL 系统列
}

// 配置
modelBuilder.Entity<Product>()
    .Property(p => p.Xmin)
    .HasColumnName("xmin")
    .HasColumnType("xid")
    .ValueGeneratedOnAddOrUpdate()
    .IsConcurrencyToken();

MySQL

// MySQL 使用 TIMESTAMP 或触发器
public class Product
{
    public int Id { get; set; }
    public string Name { get; set; }
    
    [Timestamp]
    public DateTime UpdatedAt { get; set; }
}

// 配置
modelBuilder.Entity<Product>()
    .Property(p => p.UpdatedAt)
    .HasDefaultValueSql("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP")
    .IsConcurrencyToken();

---

.NET 8/9/10 新特性

.NET 8: 改进的并发冲突诊断

builder.Services.AddDbContext<AppDbContext>(options =>
{
    options.UseSqlServer(connectionString)
           .EnableDetailedErrors()
           .LogTo(Console.WriteLine, LogLevel.Information);
});

// .NET 8 输出更详细的并发冲突信息:
// warn: Concurrency conflict detected for entity 'Product' with key values: [1]
// info: Original RowVersion: 0x00000000000007D1
// info: Database RowVersion: 0x00000000000007D2
// info: Conflicting properties: Price, Stock

.NET 9: 增强的冲突解决 API

// .NET 9 引入更便捷的冲突解决方法
catch (DbUpdateConcurrencyException ex)
{
    foreach (var entry in ex.Entries)
    {
        // 新方法: 直接获取差异
        var changes = entry.GetDatabaseValues();
        var conflicts = entry.GetConflictingProperties();
        
        foreach (var conflict in conflicts)
        {
            Console.WriteLine($"{conflict.PropertyName}:");
            Console.WriteLine($"  Original: {conflict.OriginalValue}");
            Console.WriteLine($"  Current:  {conflict.CurrentValue}");
            Console.WriteLine($"  Database: {conflict.DatabaseValue}");
        }
        
        // 新方法: 智能合并
        entry.ResolveConflicts(ConflictResolution.ClientWins);
        // 或
        entry.ResolveConflicts(ConflictResolution.StoreWins);
        // 或
        entry.ResolveConflicts(ConflictResolution.Merge);  // .NET 9 新增
    }
    
    await context.SaveChangesAsync();
}

.NET 10: 智能并发控制(路线图)

预计特性:

• 自动检测并发热点
• 基于 AI 的冲突预测
• 自动化解策略建议
• 分布式并发控制支持

---

最佳实践与陷阱

最佳实践

1. 优先使用 Timestamp

// ✅ 推荐: 使用 RowVersion
public class BaseEntity
{
    public int Id { get; set; }
    public byte[] RowVersion { get; set; }  // 所有实体继承
}

public class Product : BaseEntity { ... }
public class Order : BaseEntity { ... }

// ❌ 避免: 手动配置多个 ConcurrencyCheck
public class Product
{
    [ConcurrencyCheck]
    public decimal Price { get; set; }
    
    [ConcurrencyCheck]
    public int Stock { get; set; }
    
    [ConcurrencyCheck]
    public string Status { get; set; }
    // 💥 容易遗漏字段
}

2. 始终处理并发异常

// ✅ 推荐: 完整的异常处理
public async Task UpdateAsync(Entity entity)
{
    try
    {
        await _context.SaveChangesAsync();
    }
    catch (DbUpdateConcurrencyException ex)
    {
        _logger.LogWarning(ex, "并发冲突 detected for {EntityType}", 
            ex.Entries.First().Entity.GetType().Name);
        
        // 根据业务逻辑选择策略
        await HandleConcurrencyConflictAsync(ex);
    }
}

private async Task HandleConcurrencyConflictAsync(DbUpdateConcurrencyException ex)
{
    // 策略 1: 刷新并重试
    foreach (var entry in ex.Entries)
    {
        entry.Reload();
    }
    await _context.SaveChangesAsync();
    
    // 策略 2: 返回冲突信息给用户
    // throw new ConcurrencyConflictException("数据已被修改");
}

3. 在 API 中使用 ETag

// ✅ 推荐: 使用 HTTP ETag 标准
app.MapGet("/api/products/{id}", async (int id, AppDbContext db) =>
{
    var product = await db.Products.FindAsync(id);
    if (product == null)
        return Results.NotFound();
    
    var etag = Convert.ToBase64String(product.RowVersion);
    
    return Results.Ok(product)
        .WithETag(etag);  // 响应头: ETag: "abc123..."
});

app.MapPut("/api/products/{id}", async (
    int id, 
    ProductDto dto,
    HttpContext context,
    AppDbContext db) =>
{
    var ifMatch = context.Request.Headers.IfMatch.ToString();
    if (string.IsNullOrEmpty(ifMatch))
        return Results.BadRequest("Missing If-Match header");
    
    // 处理更新...
});

4. 单元测试并发控制

[Fact]
public async Task ConcurrentUpdates_ShouldThrowConcurrencyException()
{
    // Arrange
    var product = new Product { Name = "Test", Price = 100 };
    _context.Products.Add(product);
    await _context.SaveChangesAsync();
    
    // Act: 模拟并发更新
    var context1 = CreateDbContext();
    var context2 = CreateDbContext();
    
    var product1 = await context1.Products.FindAsync(product.Id);
    var product2 = await context2.Products.FindAsync(product.Id);
    
    product1.Price = 120;
    await context1.SaveChangesAsync();  // 第一个保存成功
    
    product2.Price = 110;
    
    // Assert
    await Assert.ThrowsAsync<DbUpdateConcurrencyException>(
        () => context2.SaveChangesAsync());
}

常见陷阱

陷阱 1: 忘记设置原始值

// ❌ 错误: 未设置原始 RowVersion
var product = await _context.Products.FindAsync(1);
product.Price = 120;
await _context.SaveChangesAsync();  // 💥 不会检测并发冲突!

// ✅ 正确: 设置原始值
var product = await _context.Products.AsNoTracking().FirstOrDefaultAsync(p => p.Id == 1);
_context.Entry(product).State = EntityState.Modified;
_context.Entry(product).OriginalValues["RowVersion"] = originalRowVersion;
product.Price = 120;
await _context.SaveChangesAsync();  // ✅ 会检测冲突

陷阱 2: AsNoTracking 与并发控制

// ⚠️ 注意: AsNoTracking 不跟踪原始值
var product = await _context.Products
    .AsNoTracking()
    .FirstOrDefaultAsync(p => p.Id == 1);

// 修改后需要手动附加
_context.Attach(product);
product.Price = 120;

// 必须设置原始 RowVersion
_context.Entry(product).OriginalValues["RowVersion"] = originalRowVersion;

await _context.SaveChangesAsync();

陷阱 3: 批量操作跳过并发检查

// ⚠️ 注意: ExecuteUpdateAsync 不检查并发
await _context.Products
    .Where(p => p.CategoryId == 1)
    .ExecuteUpdateAsync(setters => setters
        .SetProperty(p => p.Price, p => p.Price * 1.1m));

// 💥 即使有 RowVersion,也不会检查!
// 批量操作设计如此,追求性能

// ✅ 解决: 如需并发控制,逐个更新
var products = await _context.Products
    .Where(p => p.CategoryId == 1)
    .ToListAsync();

foreach (var product in products)
{
    product.Price *= 1.1m;
}

await _context.SaveChangesAsync();  // 会检查并发

---

总结

核心要点

1. 并发控制: 检测并防止多人同时修改导致的数据丢失
2. ConcurrencyCheck: 监控特定字段,灵活但需手动配置
3. Timestamp: 监控整行,自动管理,推荐使用
4. 异常处理: 捕获 DbUpdateConcurrencyException 并适当处理
5. API 设计: 使用 ETag/If-Match 头部传递 RowVersion

选择建议

场景	推荐方式	原因
SQL Server	Timestamp	原生支持,性能最优
PostgreSQL	xmin + Timestamp	系统列支持
MySQL	TIMESTAMP 列	自动更新
部分字段监控	ConcurrencyCheck	精确控制
新项目	Timestamp	简单可靠

代码模板

// 模板 1: Timestamp 实体
public class BaseEntity
{
    public int Id { get; set; }
    public byte[] RowVersion { get; set; }
}

// 模板 2: 并发更新
try
{
    _context.Entry(entity).OriginalValues["RowVersion"] = originalVersion;
    // 应用修改...
    await _context.SaveChangesAsync();
}
catch (DbUpdateConcurrencyException)
{
    // 处理冲突
    await HandleConflictAsync();
}

// 模板 3: API ETag
app.MapGet("/api/entities/{id}", async (int id, DbContext db) =>
{
    var entity = await db.Entities.FindAsync(id);
    var etag = Convert.ToBase64String(entity.RowVersion);
    return Results.Ok(entity).WithETag(etag);
});

下一步

• 📖 阅读 处理 DbUpdateConcurrencyException
• 🔧 学习 迁移生产环境最佳实践
• 🚀 了解 拦截器技术