测试 - 并发压力测试

概述

并发压力测试是验证幂等性系统正确性和性能的关键手段。通过模拟高并发场景，可以发现潜在的竞态条件、性能瓶颈和一致性问题。

测试策略

1. 测试金字塔

        /\
       /  \      E2E 测试（少量）
      /----\
     /      \    集成测试（中量）
    /--------\
   /          \  单元测试（大量）
  /------------\

2. 测试类型

类型	目标	工具
单元测试	验证单个组件逻辑	xUnit, Moq
集成测试	验证组件间交互	TestServer, Redis
并发测试	验证线程安全	Parallel, Task
压力测试	验证系统容量	k6, JMeter
混沌测试	验证容错能力	Chaos Monkey

单元测试

1. Token 服务测试

using Xunit;
using Moq;
using StackExchange.Redis;

public class TokenServiceTests
{
    [Fact]
    public async Task GenerateToken_ShouldReturnUniqueToken()
    {
        // Arrange
        var redisMock = new Mock<IConnectionMultiplexer>();
        var tokenService = new RedisIdempotencyTokenService(redisMock.Object);
        
        // Act
        var token1 = await tokenService.GenerateTokenAsync();
        var token2 = await tokenService.GenerateTokenAsync();
        
        // Assert
        Assert.NotNull(token1);
        Assert.NotNull(token2);
        Assert.NotEqual(token1, token2);
    }
    
    [Fact]
    public async Task TryAcquireToken_ShouldReturnTrue_WhenTokenIsPending()
    {
        // Arrange
        var redis = ConnectionMultiplexer.Connect("localhost");
        var db = redis.GetDatabase();
        var tokenService = new RedisIdempotencyTokenService(redis);
        var token = "test_token_123";
        
        await db.StringSetAsync($"idempotency:token:{token}", "pending");
        
        // Act
        var result = await tokenService.TryAcquireTokenAsync(token, TimeSpan.FromMinutes(30));
        
        // Assert
        Assert.True(result);
    }
    
    [Fact]
    public async Task TryAcquireToken_ShouldReturnFalse_WhenTokenAlreadyUsed()
    {
        // Arrange
        var redis = ConnectionMultiplexer.Connect("localhost");
        var db = redis.GetDatabase();
        var tokenService = new RedisIdempotencyTokenService(redis);
        var token = "test_token_123";
        
        await db.StringSetAsync($"idempotency:token:{token}", "completed");
        
        // Act
        var result = await tokenService.TryAcquireTokenAsync(token, TimeSpan.FromMinutes(30));
        
        // Assert
        Assert.False(result);
    }
}

2. 乐观锁测试

public class OptimisticLockTests
{
    [Fact]
    public async Task DeductStock_ShouldSucceed_WhenNoConflict()
    {
        // Arrange
        var dbContext = CreateInMemoryDbContext();
        var product = new Product 
        { 
            Id = Guid.NewGuid(), 
            Stock = 100, 
            Version = 0 
        };
        dbContext.Products.Add(product);
        await dbContext.SaveChangesAsync();
        
        var service = new InventoryService(dbContext);
        
        // Act
        var result = await service.DeductStockAsync(product.Id, 10);
        
        // Assert
        Assert.True(result.IsSuccess);
        
        var updatedProduct = await dbContext.Products.FindAsync(product.Id);
        Assert.Equal(90, updatedProduct.Stock);
        Assert.Equal(1, updatedProduct.Version);
    }
    
    [Fact]
    public async Task DeductStock_ShouldFail_WhenVersionMismatch()
    {
        // Arrange
        var dbContext = CreateInMemoryDbContext();
        var product = new Product 
        { 
            Id = Guid.NewGuid(), 
            Stock = 100, 
            Version = 0 
        };
        dbContext.Products.Add(product);
        await dbContext.SaveChangesAsync();
        
        var service = new InventoryService(dbContext);
        
        // Simulate version mismatch
        product.Version = 999; // Wrong version
        
        // Act
        var result = await service.DeductStockWithVersionAsync(product.Id, 10, 999);
        
        // Assert
        Assert.False(result.IsSuccess);
        Assert.Contains("version", result.Error.ToLower());
    }
}

并发测试

1. 并行请求测试

public class ConcurrentOrderTests
{
    [Fact]
    public async Task CreateOrder_ConcurrentRequests_ShouldCreateOnlyOneOrder()
    {
        // Arrange
        var serviceProvider = CreateTestServiceProvider();
        var orderService = serviceProvider.GetRequiredService<OrderService>();
        var userId = Guid.NewGuid();
        var idempotencyKey = "test_idempotency_key";
        
        var request = new CreateOrderRequest
        {
            Items = new List<OrderItemRequest>
            {
                new() { ProductId = Guid.NewGuid(), Quantity = 1, UnitPrice = 100 }
            }
        };
        
        // Act: 并发发送 10 个相同请求
        var tasks = Enumerable.Range(0, 10)
            .Select(_ => orderService.CreateOrderAsync(userId, request, idempotencyKey))
            .ToList();
        
        var results = await Task.WhenAll(tasks);
        
        // Assert
        var successfulOrders = results.Where(r => r.IsSuccess).ToList();
        var orders = successfulOrders.Select(r => r.Data).DistinctBy(o => o.Id).ToList();
        
        // 应该只有一个订单被创建
        Assert.Single(orders);
        
        // 所有成功请求应该返回同一个订单
        var orderId = orders[0].Id;
        Assert.All(successfulOrders, r => Assert.Equal(orderId, r.Data.Id));
    }
    
    [Fact]
    public async Task DeductStock_ConcurrentRequests_ShouldHandleCorrectly()
    {
        // Arrange
        var dbContext = CreateTestDbContext();
        var product = new Product 
        { 
            Id = Guid.NewGuid(), 
            Stock = 100,
            Version = 0
        };
        dbContext.Products.Add(product);
        await dbContext.SaveChangesAsync();
        
        var service = new InventoryService(dbContext);
        int successCount = 0;
        int failCount = 0;
        
        // Act: 并发扣减库存
        var tasks = Enumerable.Range(0, 20)
            .Select(async _ =>
            {
                try
                {
                    var result = await service.DeductStockAsync(product.Id, 1);
                    if (result.IsSuccess)
                    {
                        Interlocked.Increment(ref successCount);
                    }
                    else
                    {
                        Interlocked.Increment(ref failCount);
                    }
                }
                catch
                {
                    Interlocked.Increment(ref failCount);
                }
            })
            .ToList();
        
        await Task.WhenAll(tasks);
        
        // Assert
        var finalProduct = await dbContext.Products.FindAsync(product.Id);
        
        // 成功的扣减次数应该等于实际库存减少量
        Assert.Equal(100 - finalProduct.Stock, successCount);
        
        // 总请求数 = 成功 + 失败
        Assert.Equal(20, successCount + failCount);
    }
}

2. 分布式锁测试

public class DistributedLockTests
{
    [Fact]
    public async Task ExecuteWithLock_ShouldPreventConcurrentExecution()
    {
        // Arrange
        var redis = ConnectionMultiplexer.Connect("localhost");
        var lockService = new RedisDistributedLock(redis);
        var lockKey = "test_lock";
        
        var executionOrder = new ConcurrentBag<string>();
        
        // Act: 多个任务竞争同一把锁
        var tasks = Enumerable.Range(0, 5)
            .Select(async i =>
            {
                await lockService.ExecuteWithLockAsync(lockKey, async () =>
                {
                    executionOrder.Add($"Task_{i}_Start");
                    await Task.Delay(100); // 模拟业务处理
                    executionOrder.Add($"Task_{i}_End");
                    return true;
                }, TimeSpan.FromSeconds(10));
            })
            .ToList();
        
        await Task.WhenAll(tasks);
        
        // Assert
        var executions = executionOrder.ToList();
        
        // 验证没有重叠执行
        for (int i = 0; i < executions.Count; i += 2)
        {
            var start = executions[i];
            var end = executions[i + 1];
            
            var taskNum = start.Split('_')[1];
            Assert.Equal($"Task_{taskNum}_Start", start);
            Assert.Equal($"Task_{taskNum}_End", end);
        }
    }
}

压力测试

1. k6 压力测试脚本

// load_test.js
import http from 'k6/http';
import { check, sleep } from 'k6';
import { uuidv4 } from 'https://jslib.k6.io/k6-utils/1.2.0/index.js';

export const options = {
  stages: [
    { duration: '30s', target: 50 },   //  ramp up to 50 users
    { duration: '1m', target: 50 },    // stay at 50 users
    { duration: '30s', target: 100 },  // ramp up to 100 users
    { duration: '1m', target: 100 },   // stay at 100 users
    { duration: '30s', target: 0 },    // ramp down
  ],
  thresholds: {
    http_req_duration: ['p(95)<500'],  // 95% of requests should be below 500ms
    http_req_failed: ['rate<0.01'],    // error rate should be less than 1%
  },
};

export default function () {
  const idempotencyKey = uuidv4();
  
  const payload = JSON.stringify({
    userId: '123e4567-e89b-12d3-a456-426614174000',
    items: [
      {
        productId: '123e4567-e89b-12d3-a456-426614174001',
        quantity: 1,
        unitPrice: 100
      }
    ]
  });
  
  const params = {
    headers: {
      'Content-Type': 'application/json',
      'Idempotency-Key': idempotencyKey,
      'Authorization': 'Bearer test_token'
    },
  };
  
  const res = http.post('http://localhost:5000/api/orders', payload, params);
  
  check(res, {
    'status is 201 or 200': (r) => r.status === 201 || r.status === 200,
    'response has order id': (r) => JSON.parse(r.body).id !== undefined,
  });
  
  sleep(1);
}

运行测试：

k6 run load_test.js

2. C# 压力测试

public class StressTest
{
    [Fact]
    public async Task HighConcurrentLoad_ShouldMaintainConsistency()
    {
        // Arrange
        var serviceProvider = CreateTestServiceProvider();
        var orderService = serviceProvider.GetRequiredService<OrderService>();
        
        int totalRequests = 1000;
        int concurrentUsers = 50;
        int successCount = 0;
        int duplicateCount = 0;
        int errorCount = 0;
        
        var stopwatch = Stopwatch.StartNew();
        
        // Act
        var tasks = Enumerable.Range(0, totalRequests)
            .Select(async i =>
            {
                var userId = Guid.NewGuid();
                var idempotencyKey = $"stress_test_key_{i % 100}"; // 重用 100 个键
                
                var request = new CreateOrderRequest
                {
                    Items = new List<OrderItemRequest>
                    {
                        new() { ProductId = Guid.NewGuid(), Quantity = 1, UnitPrice = 100 }
                    }
                };
                
                try
                {
                    var result = await orderService.CreateOrderAsync(userId, request, idempotencyKey);
                    
                    if (result.IsSuccess)
                    {
                        Interlocked.Increment(ref successCount);
                    }
                }
                catch (Exception ex) when (ex.Message.Contains("duplicate"))
                {
                    Interlocked.Increment(ref duplicateCount);
                }
                catch
                {
                    Interlocked.Increment(ref errorCount);
                }
            });
        
        // 限制并发度
        var semaphore = new SemaphoreSlim(concurrentUsers);
        var throttledTasks = tasks.Select(async task =>
        {
            await semaphore.WaitAsync();
            try
            {
                await task;
            }
            finally
            {
                semaphore.Release();
            }
        });
        
        await Task.WhenAll(throttledTasks);
        
        stopwatch.Stop();
        
        // Assert
        Console.WriteLine($"Total Requests: {totalRequests}");
        Console.WriteLine($"Success: {successCount}");
        Console.WriteLine($"Duplicates: {duplicateCount}");
        Console.WriteLine($"Errors: {errorCount}");
        Console.WriteLine($"Duration: {stopwatch.ElapsedMilliseconds}ms");
        Console.WriteLine($"Throughput: {totalRequests / (stopwatch.ElapsedMilliseconds / 1000.0):F2} req/s");
        
        // 验证数据一致性
        Assert.Equal(0, errorCount);
        Assert.True(duplicateCount > 0); // 应该有重复检测
    }
}

监控指标

1. 收集测试指标

public class TestMetrics
{
    public int TotalRequests { get; set; }
    public int SuccessRequests { get; set; }
    public int FailedRequests { get; set; }
    public int DuplicateRequests { get; set; }
    public double AverageResponseTime { get; set; }
    public double P95ResponseTime { get; set; }
    public double P99ResponseTime { get; set; }
    public double Throughput { get; set; }
    
    public void PrintReport()
    {
        Console.WriteLine("=== Load Test Report ===");
        Console.WriteLine($"Total Requests: {TotalRequests}");
        Console.WriteLine($"Success Rate: {(double)SuccessRequests / TotalRequests * 100:F2}%");
        Console.WriteLine($"Duplicate Rate: {(double)DuplicateRequests / TotalRequests * 100:F2}%");
        Console.WriteLine($"Error Rate: {(double)FailedRequests / TotalRequests * 100:F2}%");
        Console.WriteLine($"Avg Response Time: {AverageResponseTime:F2}ms");
        Console.WriteLine($"P95 Response Time: {P95ResponseTime:F2}ms");
        Console.WriteLine($"P99 Response Time: {P99ResponseTime:F2}ms");
        Console.WriteLine($"Throughput: {Throughput:F2} req/s");
    }
}

最佳实践总结

✅ DO

1. 分层测试：单元 + 集成 + 压力测试
2. 模拟真实场景：使用生产环境的数据分布
3. 逐步增加负载：从低到高观察系统表现
4. 监控关键指标：响应时间、错误率、吞吐量
5. 自动化测试：集成到 CI/CD 流程

❌ DON'T

1. 不要只测 happy path：也要测试异常场景
2. 不要忽略预热：让系统达到稳定状态
3. 不要单次测试：多次运行取平均值
4. 不要忘记清理：测试后清理测试数据

总结

并发压力测试是保证幂等性系统质量的关键：

✅ 单元测试：验证组件逻辑
✅ 并发测试：发现竞态条件
✅ 压力测试：评估系统容量
✅ 监控指标：量化系统表现

通过全面的测试，可以确保幂等性系统在高并发场景下的正确性和稳定性。