解决几个常见的ASP.NET Core Web API 中多线程并发写入数据库失败的问题

发布于:2024-12-18 ⋅ 阅读:(58) ⋅ 点赞:(0)

在这里插入图片描述

前言

在ASP.NET Core Web API应用程序中,当多个并发线程同时调用新增用户数据的接口时,可能会遇到数据库写入失败的问题。这个问题通常源于多个线程同时访问数据库时,可能会导致以下情况:

  1. 数据库连接池耗尽:每个线程都可能创建一个数据库连接,如果并发量过大,可能会导致数据库连接池用尽,从而无法创建新的连接,导致写入失败。
  2. 数据一致性问题:多个线程同时写入数据库时,可能会造成数据冲突或违反唯一性约束。
  3. 事务问题:没有适当的事务控制,多个线程可能在执行写入时发生数据不一致或冲突。

接下来,我们将通过示例来说明如何解决这些问题。

示例:多线程并发写入数据库

为了完整地实现一个基于 ASP.NET Core Web API 的应用,使用 MySQL 数据库并处理多线程并发写入的问题,以下是一个完整的示例代码,包括了 Program.cs 中的服务注册、MySQL 配置以及其它相关的服务和依赖注入设置。

1. 配置数据库连接和服务注册

首先,确保的 appsettings.json 中包含 MySQL 的连接字符串配置:

1.1 appsettings.json
{
  "ConnectionStrings": {
    "DefaultConnection": "Server=localhost;Database=usersdb;User=root;Password=root;"
  },
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft": "Warning",
      "Microsoft.Hosting.Lifetime": "Information"
    }
  },
  "AllowedHosts": "*"
}

这个配置中的 DefaultConnection 是 MySQL 的连接字符串,确保替换为MySQL 数据库的实际连接信息。

2. 配置数据库上下文

需要使用 Entity Framework Core 来访问 MySQL 数据库,首先在 Program.cs 中注册数据库上下文。

2.1 安装 NuGet 包

在项目中安装必要的 NuGet 包,包括 Pomelo.EntityFrameworkCore.MySql(用于支持 MySQL)和 Microsoft.EntityFrameworkCore

 <ItemGroup>
   <PackageReference Include="Microsoft.EntityFrameworkCore" Version="8.0.11" />
   <PackageReference Include="Microsoft.EntityFrameworkCore.Abstractions" Version="8.0.11" />
   <PackageReference Include="Microsoft.EntityFrameworkCore.Analyzers" Version="8.0.11" />
   <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="8.0.11">
     <PrivateAssets>all</PrivateAssets>
     <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
   </PackageReference>
   <PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="8.0.11" />
   <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="8.0.11">
     <PrivateAssets>all</PrivateAssets>
     <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
   </PackageReference>
   <PackageReference Include="Pomelo.EntityFrameworkCore.MySql" Version="8.0.2" />
   <PackageReference Include="Swashbuckle.AspNetCore" Version="6.6.2" />
 </ItemGroup>

2.2 配置数据库上下文

Program.cs 中进行数据库配置,确保将 MySQL 服务注册到依赖注入容器中。

using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Sample1215.Models;
using Sample1215.Repositories;

var builder = WebApplication.CreateBuilder(args);

// 1. 注册数据库上下文服务
builder.Services.AddDbContext<ApplicationDbContext>(options =>
    options.UseMySql(builder.Configuration.GetConnectionString("DefaultConnection"), 
        ServerVersion.AutoDetect(builder.Configuration.GetConnectionString("DefaultConnection"))));

// 2. 注册自定义服务
builder.Services.AddScoped<IUserRepository, UserRepository>();

// 3. 注册控制器服务
builder.Services.AddControllers();

// 4. 注册Swagger(可选,用于API文档)
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();

var app = builder.Build();

// 5. 配置中间件
if (app.Environment.IsDevelopment())
{
    app.UseSwagger();
    app.UseSwaggerUI();
}

app.UseAuthorization();
app.MapControllers();

app.Run();

在上述代码中,我们执行了以下操作:

  • 通过 builder.Services.AddDbContext<ApplicationDbContext> 注册了数据库上下文,配置了 MySQL 数据库连接字符串。
  • 注册了 IUserRepository 接口和 UserRepository 实现,确保服务可以通过依赖注入使用。
  • 配置了Swagger(可选),以便在开发环境下自动生成API文档。

3. 创建数据库上下文和实体类

3.1 ApplicationDbContext.cs

这是数据库上下文类,继承自 DbContext,用于与 MySQL 进行交互。

using Microsoft.EntityFrameworkCore;

namespace Sample1215.Models
{
    public class ApplicationDbContext : DbContext
    {
        public ApplicationDbContext(DbContextOptions<ApplicationDbContext> options) : base(options)
        { }

        public DbSet<User> Users { get; set; }
    }
}
3.2 User.cs 实体类

这是 User 实体类,表示数据库中的用户表。

namespace Sample1215.Models
{
    public class User
    {
        public int Id { get; set; }
        public string Name { get; set; }
        public string Email { get; set; }
    }
}

4. 创建 UserRepository 实现

4.1 IUserRepository.cs

这是用户数据存储接口。

namespace Sample1215.Repositories
{
    public interface IUserRepository
    {
        Task AddUserAsync(User user);
    }
}
4.2 UserRepository.cs

这是 UserRepository 类的实现,负责将数据插入 MySQL 数据库。第一版我们这么实现

public class UserRepository : IUserRepository
{
    private readonly ApplicationDbContext _context;

    public UserRepository(ApplicationDbContext context)
    {
        _context = context;
    }

    public async Task AddUserAsync(User user)
    {
        // 模拟并发场景
        await _context.Users.AddAsync(user);
        await _context.SaveChangesAsync(); // 写入数据库
    }
}

5. 完整的 API 控制器

5.1 UsersController.cs

这是 API 控制器,负责处理用户新增请求。第一版我们这么实现。

[ApiController]
[Route("api/[controller]")]
public class UsersController : ControllerBase
{
    private readonly IUserRepository _userRepository;

    public UsersController(IUserRepository userRepository)
    {
        _userRepository = userRepository;
    }

    [HttpPost]
    public async Task<IActionResult> CreateUserAsync([FromBody] User user)
    {
        if (user == null)
        {
            return BadRequest("Invalid user data.");
        }

        try
        {
            await _userRepository.AddUserAsync(user);
            return Ok("User created successfully.");
        }
        catch (Exception ex)
        {
            return StatusCode(StatusCodes.Status500InternalServerError, ex.Message);
        }
    }
}

6. 启动和迁移数据库

6.1 在终端运行迁移命令

确保已为 ApplicationDbContext 添加了迁移并更新了数据库。在终端中运行以下命令:

dotnet ef migrations add InitialCreate
dotnet ef database update

执行完毕之后 在这里插入图片描述

这些命令将为的 MySQL 数据库创建初始的 Users 表,并将其同步到数据库中。

测试验证问题:并发写入导致失败

为了测试并发请求在 Web API 中的处理,我们可以使用单元测试框架来模拟多个并发请求。这里我们将使用 xUnit 作为单元测试框架,并使用 Microsoft.AspNetCore.Mvc.TestingHttpClient 来模拟 HTTP 请求。

1. 添加所需的 NuGet 包

在测试项目中,确保添加以下 NuGet 包:

<Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>disable</Nullable>

    <IsPackable>false</IsPackable>
    <IsTestProject>true</IsTestProject>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="coverlet.collector" Version="6.0.0" />
    <PackageReference Include="Microsoft.AspNetCore.Mvc.Testing" Version="8.0.11" />
    <PackageReference Include="Microsoft.EntityFrameworkCore.InMemory" Version="8.0.11" />
    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="8.0.1" />
    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.8.0" />
    <PackageReference Include="Moq" Version="4.20.72" />
    <PackageReference Include="xunit" Version="2.9.2" />
    <PackageReference Include="xunit.runner.visualstudio" Version="2.8.2">
      <PrivateAssets>all</PrivateAssets>
      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
    </PackageReference>
  </ItemGroup>

  <ItemGroup>
    <ProjectReference Include="..\Sample1215\Sample1215.csproj" />
  </ItemGroup>

  <ItemGroup>
    <Using Include="Xunit" />
  </ItemGroup>

</Project>

这些包将帮助我们执行 Web API 测试,模拟数据库操作,并进行并发请求测试。

2. 创建测试项目

我们的Web API 项目名称为 Sample1215,然后我们就创建一个名为 Sample1215.Test 的测试项目。在此项目中,我们将编写针对 Web API 的单元测试代码。

3. 编写并发测试代码

我们将使用 xUnit 来编写并发请求的单元测试。这个测试将模拟多个线程同时调用 CreateUserAsync API。

3.1 ConcurrencyTest.cs

创建一个 ConcurrencyTest.cs 文件来编写测试代码。

using Microsoft.AspNetCore.Mvc.Testing;
using Sample1215.Model;
using System.Net.Http.Json;

namespace Sample1215.Tests
{
    public class ConcurrencyTest : IClassFixture<WebApplicationFactory<Program>>
    {
        private readonly WebApplicationFactory<Program> _factory;

        public ConcurrencyTest(WebApplicationFactory<Program> factory)
        {
            _factory = factory;
        }

        [Fact]
        public async Task CreateUser_ConcurrentRequests_ShouldBeHandledCorrectly()
        {
            // Arrange
            var client = _factory.CreateClient();
            var user = new User
            {
                Name = "Test User",
                Email = "testuser@example.com"
            };

            // 通过多个并发请求模拟并发写入
            var tasks = new Task[20000];  // 模拟20000个并发请求

            for (int i = 0; i < tasks.Length; i++)
            {
                tasks[i] = SendPostRequest(client, user);
            }

            // Act
            await Task.WhenAll(tasks);

            // Assert
            // 你可以通过检查数据库中的记录数或检查响应状态来确保请求成功
            // 比如,检查某个唯一标识符是否插入成功,或者返回的状态码是否都为200
            // 例如,检查每个请求的状态码是否都是200
            foreach (var task in tasks)
            {
                Assert.True(task.IsCompletedSuccessfully);
            }
        }

        private async Task SendPostRequest(HttpClient client, User user)
        {
            var response = await client.PostAsJsonAsync("/api/users", user);
            response.EnsureSuccessStatusCode();  // 确保请求成功
        }
    }
}

4. 解释测试代码

  • WebApplicationFactory<Program>:这是一个 Microsoft.AspNetCore.Mvc.Testing 提供的工厂类,它允许我们在测试环境中启动 Web API,并创建 HTTP 客户端。
  • SendPostRequest:这个辅助方法负责向 Web API 发送 POST 请求,将用户数据提交到 /api/users
  • 并发请求:我们使用 Task.WhenAll 来等待多个并发请求同时执行,这模拟了多个线程同时访问 Web API 的场景。
  • Assert:在测试完成后,我们验证所有请求是否成功完成,确保所有并发请求都能被正确处理。

5. 数据库模拟

为了测试并发请求,我们使用了 InMemory 数据库来避免在真实 MySQL 数据库中进行操作。这样,所有的测试都可以在内存中完成,而不影响实际的数据库。

5.1 配置 InMemory 数据库

Program.cs 文件中,我们可以在测试中使用 InMemory 数据库,以便进行更快、更隔离的单元测试。可以将数据库上下文注册到 InMemory 提供程序中。

修改 Program.cs 中的数据库注册部分,如下所示:

// 在测试环境中使用InMemory数据库
builder.Services.AddDbContext<ApplicationDbContext>(options =>
    options.UseInMemoryDatabase("TestDatabase"));  // 使用InMemory数据库

6. 运行测试

现在,我们已经创建了一个并发测试,用于验证在 Web API 中并发请求的处理是否正确。在命令行中运行以下命令来执行单元测试:

dotnet test

7. 测试总结

我们已经实现了一个针对 POST /api/users 接口的并发请求单元测试。通过使用 xUnitWebApplicationFactory,我们可以模拟多并发请求并测试 API 在高并发场景下的稳定性。此外,使用 InMemory 数据库让我们能够快速进行测试而无需连接到真实数据库,这为开发和调试提供了便利。

在这个简单的例子中,假设API接口被多个并发线程调用。每个请求都会创建一个新的User并调用AddUserAsync方法将其插入到数据库中。如果并发线程过多,以下问题可能会发生:

  1. 数据库连接池耗尽:每个线程都需要获取数据库连接,若并发量过大,可能导致连接池用尽。
  2. 唯一性冲突:如果并发插入的用户具有相同的唯一约束(如Email),可能会出现违反唯一性约束的错误。
  3. 事务冲突:多个线程可能会同时修改相同的数据,导致事务失败。

500异常
在这里插入图片描述
和断言异常
在这里插入图片描述

解决方案

1. 使用数据库事务保证一致性

为了保证多个并发线程插入数据库时的一致性,可以使用数据库事务来确保每个写入操作都是原子的。如果有多个写入操作失败,则可以回滚事务。

public class UserRepository : IUserRepository
{
    private readonly ApplicationDbContext _context;

    public UserRepository(ApplicationDbContext context)
    {
        _context = context;
    }

    public async Task AddUserAsync(User user)
    {
        // 使用数据库事务
        using (var transaction = await _context.Database.BeginTransactionAsync())
        {
            try
            {
                await _context.Users.AddAsync(user);
                await _context.SaveChangesAsync();
                
                // 提交事务
                await transaction.CommitAsync();
            }
            catch (Exception)
            {
                // 回滚事务
                await transaction.RollbackAsync();
                throw;
            }
        }
    }
}

2. 限制并发请求

为了避免数据库连接池耗尽,可以限制API的并发请求数。我们可以通过使用SemaphoreSlim来控制并发线程的数量。

using Microsoft.AspNetCore.Mvc;
using Sample1215.Models;
using Sample1215.Repositories;

namespace Sample1215.Controllers
{
    [ApiController]
    [Route("api/[controller]")]
    public class UsersController : ControllerBase
    {
        private static SemaphoreSlim _semaphore = new SemaphoreSlim(10); // 限制最大并发10个请求
        private readonly IUserRepository _userRepository;

        public UsersController(IUserRepository userRepository)
        {
            _userRepository = userRepository;
        }

        [HttpPost]
        public async Task<IActionResult> CreateUserAsync([FromBody] User user)
        {
            if (user == null)
            {
                return BadRequest("Invalid user data.");
            }

            // 等待直到有空闲的线程
            await _semaphore.WaitAsync();

            try
            {
                await _userRepository.AddUserAsync(user);
                return Ok("User created successfully.");
            }
            catch (Exception ex)
            {
                return StatusCode(StatusCodes.Status500InternalServerError, ex.Message);
            }
            finally
            {
                // 完成后释放信号量
                _semaphore.Release();
            }
        }
    }
}

这样,最多只有10个线程可以同时写入数据库。如果更多请求到达,后续的请求会等待直到前面的请求完成。

3. 使用乐观锁解决数据冲突

如果并发写入的用户数据存在唯一性约束(例如Email),我们可以在数据库中使用乐观锁或在业务逻辑中检查唯一性。假设用户的Email字段是唯一的,插入数据之前,可以先检查数据库中是否已经存在相同的Email

public class UserRepository : IUserRepository
{
    private readonly ApplicationDbContext _context;

    public UserRepository(ApplicationDbContext context)
    {
        _context = context;
    }

    public async Task AddUserAsync(User user)
    {
        // 检查Email是否已存在
        var existingUser = await _context.Users
            .FirstOrDefaultAsync(u => u.Email == user.Email);

        if (existingUser != null)
        {
            throw new InvalidOperationException("Email already exists.");
        }

        // 如果不存在,插入新用户
        await _context.Users.AddAsync(user);
        await _context.SaveChangesAsync();
    }
}

4. 使用异步操作优化性能

异步操作有助于避免线程阻塞,从而提高API的并发处理能力。确保在数据库操作中使用异步方法。

using Microsoft.EntityFrameworkCore;
using Sample1215.Models;

namespace Sample1215.Repositories
{
    public class UserRepository : IUserRepository
    {
        private readonly ApplicationDbContext _context;

        public UserRepository(ApplicationDbContext context)
        {
            _context = context;
        }

        public async Task AddUserAsync(User user)
        {
            // 使用事务确保操作的原子性
            using (var transaction = await _context.Database.BeginTransactionAsync())
            {
                try
                {
                    // 检查Email是否已存在
                    var existingUser = await _context.Users
                        .FirstOrDefaultAsync(u => u.Email == user.Email);

                    if (existingUser != null)
                    {
                        throw new InvalidOperationException("Email already exists.");
                    }

                    await _context.Users.AddAsync(user);
                    await _context.SaveChangesAsync();

                    await transaction.CommitAsync();
                }
                catch (Exception)
                {
                    await transaction.RollbackAsync();
                    throw;
                }
            }
        }
    }
}

总结

当多个并发线程访问数据库时,可能会遇到数据库连接池耗尽、数据一致性问题以及事务冲突等问题。通过以下策略,可以有效解决这些问题:

  1. 使用数据库事务:确保每个插入操作都能原子执行,避免数据不一致。
  2. 限制并发请求:通过信号量或线程池限制并发线程数,防止连接池耗尽。
  3. 乐观锁:通过检查唯一性约束来避免并发写入冲突。
  4. 异步操作:使用异步操作提高并发性能,减少阻塞。

这些方法不仅可以提高数据库写入的稳定性,还能提升系统的整体性能和响应能力。


网站公告

今日签到

点亮在社区的每一天
去签到