并发控制的重要性

在多线程环境中访问SQLite数据库时，并发控制至关重要。不当的并发访问可能导致数据不一致、竞态条件和潜在的数据损坏。本文将详细探讨C#中SQLite的并发控制策略。

准备环境

安装 SQLite

首先，你需要在你的 C# 项目中安装 SQLite 的 NuGet 包：

`Install-Package System.Data.SQLite`

基本并发控制机制

锁机制示例

using System;
using System.Collections.Generic;
using System.Data.SQLite;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace AppLiteSql
{
    publicclass DatabaseManager
    {
        // 创建一个静态锁对象，确保线程同步
        privatestatic readonly object _lock = new object();

        // 数据库连接字符串
        privatestring _connectionString;

        public DatabaseManager(string dbPath)
        {
            _connectionString = $"Data Source={dbPath};Version=3;";
        }

        // 线程安全的插入方法
        public void ThreadSafeInsert(string name, int age)
        {
            // 使用锁确保同步
            lock (_lock)
            {
                using (var connection = new SQLiteConnection(_connectionString))
                {
                    connection.Open();
                    using (var command = new SQLiteCommand(connection))
                    {
                        command.CommandText = @"
                        INSERT INTO Users (Name, Age) 
                        VALUES (@Name, @Age)";

                        command.Parameters.AddWithValue("@Name", name);
                        command.Parameters.AddWithValue("@Age", age);

                        command.ExecuteNonQuery();
                    }
                }
            }
        }

        // 线程安全的查询方法
        public int GetUserCount()
        {
            lock (_lock)
            {
                using (var connection = new SQLiteConnection(_connectionString))
                {
                    connection.Open();
                    using (var command = new SQLiteCommand("SELECT COUNT(*) FROM Users", connection))
                    {
                        return Convert.ToInt32(command.ExecuteScalar());
                    }
                }
            }
        }
    }
}

namespace AppLiteSql
{
    internal class Program
    {
        static void Main(string[] args)
        {
            DatabaseManager dbManager = new DatabaseManager("D:\\myproject\\11Test\\AppLiteSql\\db");
            // 创建多个线程并发插入数据
            var threads = new List<Thread>();
            for (int i = 0; i < 10; i++)
            {
                int threadId = i;
                var thread = new Thread(() =>
                {
                    for (int j = 0; j < 100; j++)
                    {
                        dbManager.ThreadSafeInsert($"User_{threadId}_{j}", 30 + threadId);
                    }
                });

                threads.Add(thread);
                thread.Start();
            }

            // 等待所有线程完成
            foreach (var thread in threads)
            {
                thread.Join();
            }

            // 验证插入结果
            int totalUsers = dbManager.GetUserCount();
            Console.WriteLine($"Total Users: {totalUsers}");
        }
    }
}

高级并发控制策略

信号量控制数据库连接池

using System;
using System.Collections.Generic;
using System.Data.SQLite;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace AppLiteSql
{
    publicclass AdvancedDatabaseManager
    {
        // 使用信号量控制并发连接数  
        private readonly SemaphoreSlim _connectionSemaphore;
        privatestring _connectionString;

        public AdvancedDatabaseManager(string dbPath, int maxConcurrentConnections = 5)
        {
            _connectionString = $"Data Source={dbPath};Version=3;";
            _connectionSemaphore = new SemaphoreSlim(maxConcurrentConnections);
        }

        // 异步并发查询方法  
        public async Task<int> ConcurrentQueryAsync(string query)
        {
            await _connectionSemaphore.WaitAsync();
            try
            {
                using (var connection = new SQLiteConnection(_connectionString))
                {
                    await connection.OpenAsync();
                    using (var command = new SQLiteCommand(query, connection))
                    {
                        return Convert.ToInt32(await command.ExecuteScalarAsync());
                    }
                }
            }
            finally
            {
                _connectionSemaphore.Release();
            }
        }

        // 异步读取多行数据的方法  
        public async Task<List<User>> ReadUsersAsync(string condition = null)
        {
            await _connectionSemaphore.WaitAsync();
            try
            {
                using (var connection = new SQLiteConnection(_connectionString))
                {
                    await connection.OpenAsync();

                    string query = "SELECT Id, Name, Age FROM Users";
                    if (!string.IsNullOrEmpty(condition))
                    {
                        query += $" WHERE {condition}";
                    }

                    using (var command = new SQLiteCommand(query, connection))
                    {
                        var users = new List<User>();
                        using (var reader = await command.ExecuteReaderAsync())
                        {
                            while (await reader.ReadAsync())
                            {
                                users.Add(new User
                                {
                                    Id = reader.GetInt32(0),
                                    Name = reader.GetString(1),
                                    Age = reader.GetInt32(2)
                                });
                            }
                        }
                        return users;
                    }
                }
            }
            finally
            {
                _connectionSemaphore.Release();
            }
        }

        // 异步写入数据的方法  
        public async Task<int> WriteUserAsync(User user)
        {
            await _connectionSemaphore.WaitAsync();
            try
            {
                using (var connection = new SQLiteConnection(_connectionString))
                {
                    await connection.OpenAsync();
                    using (var command = new SQLiteCommand(connection))
                    {
                        command.CommandText = @"  
                        INSERT INTO Users (Name, Age)   
                        VALUES (@Name, @Age);  
                        SELECT last_insert_rowid();";

                        command.Parameters.AddWithValue("@Name", user.Name);
                        command.Parameters.AddWithValue("@Age", user.Age);

                        return Convert.ToInt32(await command.ExecuteScalarAsync());
                    }
                }
            }
            finally
            {
                _connectionSemaphore.Release();
            }
        }

        // 异步批量写入数据的方法  
        public async Task BulkWriteUsersAsync(List<User> users)
        {
            await _connectionSemaphore.WaitAsync();
            try
            {
                using (var connection = new SQLiteConnection(_connectionString))
                {
                    await connection.OpenAsync();
                    using (var transaction = connection.BeginTransaction())
                    {
                        try
                        {
                            using (var command = new SQLiteCommand(connection))
                            {
                                command.CommandText = @"  
                                INSERT INTO Users (Name, Age)   
                                VALUES (@Name, @Age)";

                                var nameParam = command.Parameters.Add("@Name", System.Data.DbType.String);
                                var ageParam = command.Parameters.Add("@Age", System.Data.DbType.Int32);

                                foreach (var user in users)
                                {
                                    nameParam.Value = user.Name;
                                    ageParam.Value = user.Age;
                                    await command.ExecuteNonQueryAsync();
                                }
                            }

                            await transaction.CommitAsync();
                        }
                        catch
                        {
                            await transaction.RollbackAsync();
                            throw;
                        }
                    }
                }
            }
            finally
            {
                _connectionSemaphore.Release();
            }
        }

        // 异步更新数据的方法  
        public async Task<int> UpdateUserAsync(int id, User updatedUser)
        {
            await _connectionSemaphore.WaitAsync();
            try
            {
                using (var connection = new SQLiteConnection(_connectionString))
                {
                    await connection.OpenAsync();
                    using (var command = new SQLiteCommand(connection))
                    {
                        command.CommandText = @"  
                        UPDATE Users   
                        SET Name = @Name, Age = @Age   
                        WHERE Id = @Id";

                        command.Parameters.AddWithValue("@Name", updatedUser.Name);
                        command.Parameters.AddWithValue("@Age", updatedUser.Age);
                        command.Parameters.AddWithValue("@Id", id);

                        return await command.ExecuteNonQueryAsync();
                    }
                }
            }
            finally
            {
                _connectionSemaphore.Release();
            }
        }
    }

}

namespace AppLiteSql
{
    internal class Program
    {
        static async Task Main(string[] args)
        {
            AdvancedDatabaseManager dbManager = new AdvancedDatabaseManager("D:\\myproject\\11Test\\AppLiteSql\\db");
            // 写入单个用户  
            var newUser = new User { Name = "John Doe", Age = 30 };
            int newUserId = await dbManager.WriteUserAsync(newUser);

            // 批量写入用户  
            var userList = new List<User>
            {
                new User { Name = "Alice", Age = 25 },
                new User { Name = "Bob", Age = 35 }
            };
            await dbManager.BulkWriteUsersAsync(userList);

            // 读取用户  
            var users = await dbManager.ReadUsersAsync("Age > 20");
            foreach (var user in users)
            {
                Console.WriteLine($"User: {user.Name}, Age: {user.Age}");
            }

            // 更新用户  
            var updatedUser = new User { Name = "John Smith", Age = 31 };
            await dbManager.UpdateUserAsync(newUserId, updatedUser);
        }
    }
}

最佳实践与注意事项

1. 始终使用参数化查询防止SQL注入
2. 尽量缩小锁的作用范围
3. 考虑使用异步方法处理数据库操作
4. 对于高并发场景，考虑使用连接池
5. 定期检查和优化数据库性能

性能建议

• 对于读多写少的场景，考虑使用读写锁
• 使用批量插入减少数据库连接开销
• 优化查询语句和索引

总结

SQLite的并发控制需要谨慎处理。通过合理的锁机制、信号量控制和异步编程，可以有效管理多线程环境下的数据库访问。关键在于平衡线程安全性和性能。

希望这篇文章能帮助您深入理解C#中SQLite的并发控制与多线程访问。建议根据具体业务场景选择最适合的并发策略。

阅读原文：原文链接