1. 库存并发测试的核心目标

库存并发测试主要验证在高并发场景下：

• 库存数据的准确性和一致性（不超卖不少卖）
• 系统的高可用性和稳定性
• 业务规则的正确执行（如限购策略）
• 系统异常情况下的正确处理

2. 测试场景设计

2.1 基础并发测试场景

• 普通商品库存扣减：模拟多用户同时购买同一商品
• 秒杀商品库存扣减：极端高并发场景
• 购物车批量下单：涉及多个商品库存同时扣减

2.2 异常场景

• 库存不足时的并发请求
• 重复提交订单
• 支付超时后库存回滚
• 分布式系统网络分区情况

3. 测试实现方案

3.1 使用JMeter实现

<!-- JMeter测试计划示例 -->
<ThreadGroup guiclass="ThreadGroupGui" testclass="ThreadGroup" testname="库存并发测试" enabled="true">
  <intProp name="ThreadGroup.num_threads">1000</intProp>
  <intProp name="ThreadGroup.ramp_time">10</intProp>
  <longProp name="ThreadGroup.start_time">0</longProp>
  <longProp name="ThreadGroup.end_time">0</longProp>
  <boolProp name="ThreadGroup.scheduler">false</boolProp>
</ThreadGroup>

<HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="扣减库存请求" enabled="true">
  <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
    <collectionProp name="Arguments.arguments">
      <elementProp name="" elementType="HTTPArgument">
        <stringProp name="Argument.value">{"productId":123,"quantity":1}</stringProp>
        <stringProp name="Argument.metadata">=</stringProp>
      </elementProp>
    </collectionProp>
  </elementProp>
  <stringProp name="HTTPSampler.domain">api.mall.com</stringProp>
  <stringProp name="HTTPSampler.port">443</stringProp>
  <stringProp name="HTTPSampler.protocol">https</stringProp>
  <stringProp name="HTTPSampler.path">/inventory/deduct</stringProp>
  <stringProp name="HTTPSampler.method">POST</stringProp>
</HTTPSamplerProxy>

3.2 使用Python+Locust实现

from locust import HttpUser, task, between
import random

class InventoryUser(HttpUser):
    wait_time = between(0.1, 0.5)
    
    @task
    def deduct_inventory(self):
        product_id = random.randint(1, 10)  # 测试10个商品
        self.client.post(
            "/inventory/deduct",
            json={"productId": product_id, "quantity": 1},
            headers={"Authorization": "Bearer token"}
        )
    
    def on_start(self):
        # 登录获取token
        response = self.client.post("/login", json={"username": "test", "password": "test"})
        self.token = response.json()["token"]

3.3 使用Java+TestNG实现

import org.testng.annotations.Test;
import io.restassured.RestAssured;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class InventoryConcurrencyTest {
    private static final int THREAD_COUNT = 100;
    private static final String API_URL = "http://api.mall.com/inventory/deduct";

    @Test
    public void testConcurrentDeduction() throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
        
        for (int i = 0; i < THREAD_COUNT; i++) {
            executor.execute(() -> {
                String requestBody = "{\"productId\":123,\"quantity\":1}";
                
                RestAssured.given()
                    .contentType("application/json")
                    .body(requestBody)
                    .post(API_URL)
                    .then()
                    .statusCode(200);
            });
        }
        
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
        
        // 验证最终库存
        RestAssured.given()
            .get("http://api.mall.com/inventory/123")
            .then()
            .statusCode(200)
            .body("stock", equalTo(EXPECTED_STOCK));
    }
}

4. 验证方法

4.1 数据库层面验证

-- 测试前记录初始库存
SELECT stock FROM products WHERE id = 123;

-- 测试后验证
SELECT 
    stock AS current_stock,
    (SELECT stock FROM products_before_test WHERE id = 123) - sold_quantity AS expected_stock
FROM products WHERE id = 123;

4.2 业务层面验证

1. 检查订单数量与库存减少量是否一致
2. 检查是否有超卖现象（库存不为负）
3. 检查是否有少卖现象（实际库存>预期库存）

4.3 日志分析

• 检查是否有死锁或长时间等待日志
• 检查重试机制是否正常工作
• 检查异常处理日志

5. 常见问题及解决方案

5.1 超卖问题

解决方案验证：

• 悲观锁：SELECT FOR UPDATE
• 乐观锁：version字段
• Redis分布式锁
• 库存预扣减（支付成功再实际扣减）

测试用例：

def test_inventory_over_sell():
    initial_stock = get_stock(123)
    threads = []
    
    # 模拟并发请求数超过库存
    for _ in range(initial_stock + 10):
        t = threading.Thread(target=deduct_inventory, args=(123, 1))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    final_stock = get_stock(123)
    assert final_stock >= 0  # 库存不能为负
    orders = get_order_count(123)
    assert orders <= initial_stock  # 订单数不超过初始库存

5.2 性能瓶颈

监控指标：

• TPS（每秒处理事务数）
• 平均响应时间
• 错误率
• 数据库连接池使用率
• CPU和内存使用率

优化验证：

// 使用JMeter进行梯度压测
SteppingThreadGroup:
- Start with 100 threads
- Increment by 100 every 30 seconds
- Maximum 1000 threads

// 监控各项指标随并发数增加的变化

6. 高级测试场景

6.1 分布式锁测试

def test_distributed_lock():
    # 模拟多个服务实例
    services = [InventoryService() for _ in range(3)]
    
    # 并发扣减库存
    with ThreadPoolExecutor(max_workers=100) as executor:
        futures = [executor.submit(lambda s: s.deduct(123, 1), random.choice(services)) 
                  for _ in range(200)]
        
        for future in as_completed(futures):
            assert future.result().status_code in [200, 429]  # 429表示被限流
    
    # 验证最终库存
    assert get_stock(123) == EXPECTED_STOCK

6.2 库存回滚测试

@Test
public void testInventoryRollback() {
    // 1. 扣减库存
    Response deductResponse = given()
        .body("{ \"productId\": 123, \"quantity\": 1 }")
        .post("/inventory/deduct");
    
    String orderId = deductResponse.jsonPath().getString("orderId");
    
    // 2. 模拟支付超时
    wait(30, MINUTES);
    
    // 3. 验证库存是否回滚
    int finalStock = given()
        .get("/inventory/123")
        .jsonPath().getInt("stock");
    
    assertEquals(INITIAL_STOCK, finalStock);
    
    // 4. 验证订单状态
    String orderStatus = given()
        .get("/orders/" + orderId)
        .jsonPath().getString("status");
    
    assertEquals("CANCELLED", orderStatus);
}

7. 测试报告关键指标

1. 库存准确性：预期减少量 vs 实际减少量超卖发生次数
2. 系统性能：不同并发量下的TPS响应时间百分位（P90/P95/P99）错误率趋势
3. 资源使用：数据库CPU使用率应用服务器负载连接池使用情况
4. 业务指标：成功订单数失败订单分类统计库存流水记录完整性

通过以上全面的库存并发接口测试，可以确保商城系统在高并发场景下库存管理的正确性和系统的稳定性。