EntityFramework多线程安全问题探究

Posted 2021-12-10 シ゛甜虾

一、多线程操作数据库出现的问题

因为EntityFramework的DataContext不是线程安全的，所有多线程使用EntityFramework的DataContext遇到了下面错误

“A second operation was started on this context before a previous operation completed. This is usually caused by different threads concurrently using the same instance of DbContext. For more information on how to avoid threading issues with DbContext, see https://go.microsoft.com/fwlink/?linkid=2097913.”

即使使用原生的mysqlConnection对象，多线程访问同样会出现问题
测试代码

using DependencyInjection;
using EntityDbService.Models;
using EntityDbService.Table;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Share.SystemParameter;

namespace DataBaseTest
{
    public class Program
    {
        public static void Main(string[] args)
        {
            var serviceProvider = new ServiceCollection()
                    .AddDbContext<DataContext>(optionsBuilder => optionsBuilder.UseMySql(Parameters.GetConnectionString(), new MySqlServerVersion(new Version(5, 7, 0))))
                    .BuildServiceProvider();
            SingleService.SetServiceProvider(serviceProvider);
            StartEntityTest();
        }

        private static List<Thread> _threads = new List<Thread>();
        private static bool _isStopThread = false;

        public static void StartEntityTest()
        {
            Thread addThread1 = new Thread(AddThread);
            _threads.Add(addThread1);
            addThread1.Start();

            Thread updateThread1 = new Thread(UpdateThread);
            _threads.Add(updateThread1);
            updateThread1.Start();

            Thread selectThread1 = new Thread(SelectThread);
            _threads.Add(selectThread1);
            selectThread1.Start();
        }

        public static void AddThread()
        {
            var dataContext = SingleService.Services.GetService<DataContext>();
            while (!_isStopThread)
            {
                try
                {
                    EquipmentTable equipmentTable = new EquipmentTable();
                    equipmentTable.Name = "test";
                    equipmentTable.ImeiStr = "123456789";
                    equipmentTable.Timestamp = DateTime.Now;
                    equipmentTable.CompanyId = 0;
                    dataContext.EquipmentTables.Add(equipmentTable);
                    var count = dataContext.SaveChanges();
                    Console.WriteLine($"AddThread:{Thread.GetCurrentProcessorId},equipmentTable.SaveChanges():{count}");
                    Thread.Sleep(100);
                }
                catch(Exception ex)
                {
                    Console.WriteLine($"AddThread:{Thread.GetCurrentProcessorId},Exception:{ex.ToString()}");
                    Thread.Sleep(1000);
                }
            }
        }

        public static void UpdateThread()
        {
            var dataContext = SingleService.Services.GetService<DataContext>();
            while (!_isStopThread)
            {
                try
                {
                    Random random = new Random();
                    var id = random.Next(1, 10000);
                    var equipmentTable = dataContext.EquipmentTables.Find(id);
                    if (equipmentTable != null)
                    {
                        equipmentTable.CompanyId = equipmentTable.CompanyId + 1;
                        var count = dataContext.SaveChanges();
                        Console.WriteLine($"UpdateThread:{Thread.GetCurrentProcessorId},equipmentTable.SaveChanges():{count}");
                    }
                    else
                    {
                        Console.WriteLine($"UpdateThread:{Thread.GetCurrentProcessorId},equipmentTable.SaveChanges():{0}");
                    }
                    Thread.Sleep(100);
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"UpdateThread:{Thread.GetCurrentProcessorId},Exception:{ex.ToString()}");
                    Thread.Sleep(1000);
                }
            }
        }

        public static void SelectThread()
        {
            var dataContext = SingleService.Services.GetService<DataContext>();
            while (!_isStopThread)
            {
                try
                {
                    Random random = new Random();
                    var id = random.Next(1, 10000);
                    var equipmentTable = dataContext.EquipmentTables.Skip(id).Take(id * 10);
                    if (equipmentTable != null && equipmentTable.Count() > 0)
                    {
                        Console.WriteLine($"SelectThread:{Thread.GetCurrentProcessorId},equipmentTable.Count():{equipmentTable.Count()}");
                    }
                    else
                    {
                        Console.WriteLine($"SelectThread:{Thread.GetCurrentProcessorId},equipmentTable.Count():{0}");
                    }
                    Thread.Sleep(100);
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"SelectThread:{Thread.GetCurrentProcessorId},Exception:{ex.ToString()}");
                    Thread.Sleep(1000);
                }
            }
        }
    }
}

二、第一个想法，为每个数据表创建一个DataContext，然后加锁

创建多个DataContext

.AddDbContext<DataContext>(optionsBuilder => optionsBuilder.UseMySql(ConnectionString, new MySqlServerVersion(new Version(5, 7, 0))))
.AddDbContext<EquipmentContext>(optionsBuilder => optionsBuilder.UseMySql(ConnectionString, new MySqlServerVersion(new Version(5, 7, 0))))

很显然是不可行的

PM> Add-Migration init
Build started...
Build succeeded.
More than one DbContext was found. Specify which one to use. Use the '-Context' parameter for PowerShell commands and the '--context' parameter for dotnet commands.

三、第二个想法，全局锁，很显然可以用，但是性能不行

测试代码

using EntityDbService.Table;
using Microsoft.EntityFrameworkCore;

namespace EntityDbService.Models
{
    public class DataContext : DbContext
    {

        public DataContext(DbContextOptions<DataContext> options)
            : base(options)
        {

            if (Database.GetPendingMigrations().Any())
            {
                Database.Migrate(); //执行迁移
            }
        }

        public DbSet<EquipmentTable> EquipmentTables { get; set; }
        public object Lock { get => _sign; set => _sign = value; }
        //锁
        private object _sign = new object();
    }
}

使用时要全部加锁

using DependencyInjection;
using EntityDbService.Models;
using EntityDbService.Table;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace DataBaseTest
{
    public static class EntityTest
    {
        private static List<Thread> _threads = new List<Thread>();
        private static bool _isStopThread = false;
        private static int _time = 10;

        public static void StartEntityTest()
        {
            Thread addThread1 = new Thread(AddThread);
            _threads.Add(addThread1);
            addThread1.Start();

            Thread addThread2 = new Thread(AddThread);
            _threads.Add(addThread2);
            addThread2.Start();

            Thread updateThread1 = new Thread(UpdateThread);
            _threads.Add(updateThread1);
            updateThread1.Start();

            Thread updateThread2 = new Thread(UpdateThread);
            _threads.Add(updateThread2);
            updateThread2.Start();

            Thread selectThread1 = new Thread(SelectThread);
            _threads.Add(selectThread1);
            selectThread1.Start();

            Thread selectThread2 = new Thread(SelectThread);
            _threads.Add(selectThread2);
            selectThread2.Start();
        }

        public static void AddThread()
        {
            var dataContext = SingleService.Services.GetService<DataContext>();
            while (!_isStopThread)
            {
                try
                {
                    int count = 0;
                    lock (dataContext.Lock)
                    {
                        EquipmentTable equipmentTable = new EquipmentTable();
                        equipmentTable.Name = "test";
                        equipmentTable.ImeiStr = "123456789";
                        equipmentTable.Timestamp = DateTime.Now;
                        equipmentTable.CompanyId = 0;
                        dataContext.EquipmentTables.Add(equipmentTable);
                        count = dataContext.SaveChanges();
                    }
                    Console.WriteLine($"AddThread:{Thread.GetCurrentProcessorId()},equipmentTable.SaveChanges():{count}");
                    Thread.Sleep(_time);
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"AddThread:{Thread.GetCurrentProcessorId()},Exception:{ex.ToString()}");
                    Thread.Sleep(1000);
                }
            }
        }

        public static void UpdateThread()
        {
            var dataContext = SingleService.Services.GetService<DataContext>();
            while (!_isStopThread)
            {
                try
                {
                    Random random = new Random();
                    var id = random.Next(1, 10000);
                    var count = 0;
                    lock (dataContext.Lock)
                    {
                        EquipmentTable equipmentTable = dataContext.EquipmentTables.Find(id);
                        if (equipmentTable != null)
                        {
                            equipmentTable.CompanyId = equipmentTable.CompanyId + 1;
                            count = dataContext.SaveChanges();
                        }
                    }
                    Console.WriteLine($"UpdateThread:{Thread.GetCurrentProcessorId()},equipmentTable.SaveChanges():{count}");
                    Thread.Sleep(_time);
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"UpdateThread:{Thread.GetCurrentProcessorId()},Exception:{ex.ToString()}");
                    Thread.Sleep(1000);
                }
            }
        }

        public static void SelectThread()
        {
            var dataContext = SingleService.Services.GetService<DataContext>();
            while (!_isStopThread)
            {
                try
                {
                    Random random = new Random();
                    var id = random.Next(1, 10000);
                    lock (dataContext.Lock)
                    {
                        var equipmentTable = dataContext.EquipmentTables.Skip(id).Take(10);
                   
                        if (equipmentTable != null && equipmentTable.Count() > 0)
                        {
                            Console.WriteLine($"SelectThread:{Thread.GetCurrentProcessorId()},equipmentTable.Count():{equipmentTable.Count()}");
                        }
                        else
                        {
                            Console.WriteLine($"SelectThread:{Thread.GetCurrentProcessorId()},equipmentTable.Count():{0}");
                        }
                    }
                    Thread.Sleep(_time);
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"SelectThread:{Thread.GetCurrentProcessorId()},Exception:{ex.ToString()}");
                    Thread.Sleep(1000);
                }
            }
        }
    }
}

四、第三个想法，建立一个数据库连接池

池中每个实例都是一个独立的数据库连接，每次要使用时都去池中请求空闲连接，然后加锁使用，因为连接多所以性能较单个数据库连接加锁要好很多，如果都没有空闲，可以通过机制进行等待或者增加连接池中连接。