线程安全
知识点
什么是线程安全
线程安全是指在多线程环境下，代码能够正确处理多个线程同时访问共享资源的情况，保证程序的行为是可预测和正确的。线程安全的代码在并发执行时不会出现数据竞争、状态不一致或其他并发问题。
线程安全的层次
不可变（Immutable）：对象创建后状态不能改变
线程安全（Thread-Safe）：可以安全地被多个线程同时访问
有条件线程安全（Conditionally Thread-Safe）：在特定条件下线程安全
非线程安全（Not Thread-Safe）：需要外部同步机制
常见的线程安全问题
数据竞争（Data Race）：多个线程同时访问同一内存位置
竞态条件（Race Condition）：程序结果依赖于线程执行的时序
死锁（Deadlock）：多个线程相互等待
活锁（Livelock）：线程持续重试但无法进展
饥饿（Starvation）：某些线程长期得不到执行
实现线程安全的方法
同步机制：lock、Monitor、Mutex等
原子操作：Interlocked类
不可变对象：只读属性和字段
线程本地存储：ThreadLocal
并发集合：ConcurrentDictionary、ConcurrentQueue等
代码案例
案例1：线程安全的单例模式
using System;
using System.Threading;
using System.Threading.Tasks;
// 方式1：双重检查锁定
public sealed class ThreadSafeSingleton
{
private static ThreadSafeSingleton instance = null;
private static readonly object lockObject = new object();
private ThreadSafeSingleton()
{
Console.WriteLine($"单例实例创建 - 线程ID: {Thread.CurrentThread.ManagedThreadId}");
// 模拟初始化工作
Thread.Sleep(100);
}
public static ThreadSafeSingleton Instance
{
get
{
// 第一次检查（无锁）
if (instance == null)
{
lock (lockObject)
{
// 第二次检查（有锁）
if (instance == null)
{
instance = new ThreadSafeSingleton();
}
}
}
return instance;
}
}
public void DoSomething()
{
Console.WriteLine($"执行操作 - 线程ID: {Thread.CurrentThread.ManagedThreadId}");
}
}
// 方式2：懒加载（推荐）
public sealed class LazySingleton
{
private static readonly Lazy
new Lazy
private LazySingleton()
{
Console.WriteLine($"懒加载单例创建 - 线程ID: {Thread.CurrentThread.ManagedThreadId}");
Thread.Sleep(100);
}
public static LazySingleton Instance => lazy.Value;
public void DoSomething()
{
Console.WriteLine($"懒加载单例操作 - 线程ID: {Thread.CurrentThread.ManagedThreadId}");
}
}
class SingletonExample
{
static void Main(string[] args)
{
Console.WriteLine("线程安全单例模式测试");
// 测试双重检查锁定单例
Console.WriteLine("\n=== 双重检查锁定单例 ===");
Task[] tasks1 = new Task[5];
for (int i = 0; i < tasks1.Length; i++)
{
tasks1[i] = Task.Run(() =>
{
var singleton = ThreadSafeSingleton.Instance;
singleton.DoSomething();
});
}
Task.WaitAll(tasks1);
// 测试懒加载单例
Console.WriteLine("\n=== 懒加载单例 ===");
Task[] tasks2 = new Task[5];
for (int i = 0; i < tasks2.Length; i++)
{
tasks2[i] = Task.Run(() =>
{
var singleton = LazySingleton.Instance;
singleton.DoSomething();
});
}
Task.WaitAll(tasks2);
Console.WriteLine("单例测试完成");
}
}
案例2：线程安全的计数器
using System;
using System.Threading;
using System.Threading.Tasks;
// 非线程安全的计数器
public class UnsafeCounter
{
private int count = 0;
public void Increment()
{
count++; // 非原子操作
}
public int Value => count;
}
// 使用锁的线程安全计数器
public class LockBasedCounter
{
private int count = 0;
private readonly object lockObject = new object();
public void Increment()
{
lock (lockObject)
{
count++;
}
}
public int Value
{
get
{
lock (lockObject)
{
return count;
}
}
}
}
// 使用原子操作的线程安全计数器
public class AtomicCounter
{
private int count = 0;
public void Increment()
{
Interlocked.Increment(ref count);
}
public int Value => Volatile.Read(ref count);
}
// 高级线程安全计数器
public class AdvancedCounter
{
private long count = 0;
private long operations = 0;
private readonly object lockObject = new object();
public void Increment()
{
Interlocked.Increment(ref count);
Interlocked.Increment(ref operations);
}
public void Decrement()
{
Interlocked.Decrement(ref count);
Interlocked.Increment(ref operations);
}
public void Add(int value)
{
Interlocked.Add(ref count, value);
Interlocked.Increment(ref operations);
}
public long Value => Volatile.Read(ref count);
public long Operations => Volatile.Read(ref operations);
public void Reset()
{
lock (lockObject)
{
Volatile.Write(ref count, 0);
Volatile.Write(ref operations, 0);
}
}
}
class CounterExample
{
static void Main(string[] args)
{
Console.WriteLine("线程安全计数器对比测试");
const int tasksCount = 10;
const int incrementsPerTask = 10000;
const int expectedTotal = tasksCount * incrementsPerTask;
// 测试非线程安全计数器
TestCounter("非线程安全计数器",
() => new UnsafeCounter(),
counter => counter.Increment(),
counter => counter.Value,
tasksCount, incrementsPerTask, expectedTotal);
// 测试基于锁的计数器
TestCounter("基于锁的计数器",
() => new LockBasedCounter(),
counter => counter.Increment(),
counter => counter.Value,
tasksCount, incrementsPerTask, expectedTotal);
// 测试原子操作计数器
TestCounter("原子操作计数器",
() => new AtomicCounter(),
counter => counter.Increment(),
counter => counter.Value,
tasksCount, incrementsPerTask, expectedTotal);
// 测试高级计数器
Console.WriteLine("\n=== 高级计数器功能测试 ===");
var advancedCounter = new AdvancedCounter();
Task[] advancedTasks = new Task[6];
// 递增任务
for (int i = 0; i < 3; i++)
{
advancedTasks[i] = Task.Run(() =>
{
for (int j = 0; j < 1000; j++)
{
advancedCounter.Increment();
}
});
}
// 递减任务
advancedTasks[3] = Task.Run(() =>
{
for (int j = 0; j < 500; j++)
{
advancedCounter.Decrement();
}
});
// 批量增加任务
advancedTasks[4] = Task.Run(() =>
{
for (int j = 0; j < 100; j++)
{
advancedCounter.Add(5);
}
});
// 批量减少任务
advancedTasks[5] = Task.Run(() =>
{
for (int j = 0; j < 50; j++)
{
advancedCounter.Add(-10);
}
});
Task.WaitAll(advancedTasks);
Console.WriteLine($"高级计数器最终值: {advancedCounter.Value}");
Console.WriteLine($"总操作次数: {advancedCounter.Operations}");
// 期望值计算: 3000(递增) - 500(递减) + 500(批量增加) - 500(批量减少) = 2500
Console.WriteLine($"期望值: 2500");
}
static void TestCounter
Func
{
Console.WriteLine($"\n=== {name} ===");
var counter = factory();
var tasks = new Task[tasksCount];
var stopwatch = System.Diagnostics.Stopwatch.StartNew();
for (int i = 0; i < tasksCount; i++)
{
tasks[i] = Task.Run(() =>
{
for (int j = 0; j < incrementsPerTask; j++)
{
increment(counter);
}
});
}
Task.WaitAll(tasks);
stopwatch.Stop();
int finalValue = getValue();
Console.WriteLine($"期望值: {expectedTotal}");
Console.WriteLine($"实际值: {finalValue}");
Console.WriteLine($"正确性: {(finalValue == expectedTotal ? "✓" : "✗")}");
Console.WriteLine($"耗时: {stopwatch.ElapsedMilliseconds} ms");
Console.WriteLine($"吞吐量: {(tasksCount * incrementsPerTask * 1000.0 / stopwatch.ElapsedMilliseconds):F0} ops/sec");
}
}
案例3：线程安全的缓存实现
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
public interface ICache
{
TValue Get(TKey key);
void Set(TKey key, TValue value);
bool TryGet(TKey key, out TValue value);
void Remove(TKey key);
void Clear();
int Count { get; }
}
// 使用读写锁的缓存实现
public class ReaderWriterCache
{
private readonly Dictionary
private readonly ReaderWriterLockSlim rwLock = new ReaderWriterLockSlim();
private readonly TimeSpan defaultExpiry;
private class CacheItem
{
public TValue Value { get; set; }
public DateTime ExpiryTime { get; set; }
public bool IsExpired => DateTime.UtcNow > ExpiryTime;
}
public ReaderWriterCache(TimeSpan defaultExpiry)
{
this.defaultExpiry = defaultExpiry;
}
public TValue Get(TKey key)
{
rwLock.EnterReadLock();
try
{
if (cache.TryGetValue(key, out CacheItem item) && !item.IsExpired)
{
return item.Value;
}
throw new KeyNotFoundException($"Key '{key}' not found or expired");
}
finally
{
rwLock.ExitReadLock();
}
}
public bool TryGet(TKey key, out TValue value)
{
rwLock.EnterReadLock();
try
{
if (cache.TryGetValue(key, out CacheItem item) && !item.IsExpired)
{
value = item.Value;
return true;
}
value = default(TValue);
return false;
}
finally
{
rwLock.ExitReadLock();
}
}
public void Set(TKey key, TValue value)
{
rwLock.EnterWriteLock();
try
{
cache[key] = new CacheItem
{
Value = value,
ExpiryTime = DateTime.UtcNow.Add(defaultExpiry)
};
}
finally
{
rwLock.ExitWriteLock();
}
}
public void Remove(TKey key)
{
rwLock.EnterWriteLock();
try
{
cache.Remove(key);
}
finally
{
rwLock.ExitWriteLock();
}
}
public void Clear()
{
rwLock.EnterWriteLock();
try
{
cache.Clear();
}
finally
{
rwLock.ExitWriteLock();
}
}
public int Count
{
get
{
rwLock.EnterReadLock();
try
{
return cache.Count;
}
finally
{
rwLock.ExitReadLock();
}
}
}
public void Dispose()
{
rwLock?.Dispose();
}
}
// 使用并发集合的缓存实现
public class ConcurrentCache
{
private readonly ConcurrentDictionary
new ConcurrentDictionary
private readonly TimeSpan defaultExpiry;
private readonly Timer cleanupTimer;
private class CacheItem
{
public TValue Value { get; set; }
public DateTime ExpiryTime { get; set; }
public bool IsExpired => DateTime.UtcNow > ExpiryTime;
}
public ConcurrentCache(TimeSpan defaultExpiry)
{
this.defaultExpiry = defaultExpiry;
// 每10秒清理一次过期项
cleanupTimer = new Timer(CleanupExpiredItems, null,
TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(10));
}
public TValue Get(TKey key)
{
if (TryGet(key, out TValue value))
{
return value;
}
throw new KeyNotFoundException($"Key '{key}' not found or expired");
}
public bool TryGet(TKey key, out TValue value)
{
if (cache.TryGetValue(key, out CacheItem item) && !item.IsExpired)
{
value = item.Value;
return true;
}
// 如果过期，尝试移除
if (item != null && item.IsExpired)
{
cache.TryRemove(key, out _);
}
value = default(TValue);
return false;
}
public void Set(TKey key, TValue value)
{
cache.AddOrUpdate(key,
new CacheItem
{
Value = value,
ExpiryTime = DateTime.UtcNow.Add(defaultExpiry)
},
(k, oldItem) => new CacheItem
{
Value = value,
ExpiryTime = DateTime.UtcNow.Add(defaultExpiry)
});
}
public void Remove(TKey key)
{
cache.TryRemove(key, out _);
}
public void Clear()
{
cache.Clear();
}
public int Count => cache.Count;
private void CleanupExpiredItems(object state)
{
var keysToRemove = new List
foreach (var kvp in cache)
{
if (kvp.Value.IsExpired)
{
keysToRemove.Add(kvp.Key);
}
}
foreach (var key in keysToRemove)
{
cache.TryRemove(key, out _);
}
if (keysToRemove.Count > 0)
{
Console.WriteLine($"清理了 {keysToRemove.Count} 个过期缓存项");
}
}
}
class CacheExample
{
static void Main(string[] args)
{
Console.WriteLine("线程安全缓存测试");
TimeSpan expiry = TimeSpan.FromSeconds(5);
// 测试读写锁缓存
Console.WriteLine("\n=== 读写锁缓存测试 ===");
using (var rwCache = new ReaderWriterCache
{
TestCache(rwCache, "RWLockCache");
}
// 测试并发集合缓存
Console.WriteLine("\n=== 并发集合缓存测试 ===");
var concurrentCache = new ConcurrentCache
TestCache(concurrentCache, "ConcurrentCache");
Console.WriteLine("缓存测试完成");
}
static void TestCache(ICache
{
const int tasksCount = 8;
const int operationsPerTask = 1000;
var tasks = new Task[tasksCount];
var random = new Random();
for (int i = 0; i < tasksCount; i++)
{
int taskId = i;
tasks[i] = Task.Run(() =>
{
var taskRandom = new Random(taskId);
for (int j = 0; j < operationsPerTask; j++)
{
string key = $"key_{taskRandom.Next(1, 100)}";
int operation = taskRandom.Next(1, 101);
if (operation <= 70) // 70% 读操作
{
if (cache.TryGet(key, out string value))
{
// 读取成功
}
}
else if (operation <= 95) // 25% 写操作
{
string value = $"value_{taskId}_{j}_{DateTime.Now:HH:mm:ss.fff}";
cache.Set(key, value);
}
else // 5% 删除操作
{
cache.Remove(key);
}
}
});
}
var stopwatch = System.Diagnostics.Stopwatch.StartNew();
Task.WaitAll(tasks);
stopwatch.Stop();
Console.WriteLine($"{cacheName} - 缓存项数量: {cache.Count}");
Console.WriteLine($"{cacheName} - 总耗时: {stopwatch.ElapsedMilliseconds} ms");
Console.WriteLine($"{cacheName} - 吞吐量: {(tasksCount * operationsPerTask * 1000.0 / stopwatch.ElapsedMilliseconds):F0} ops/sec");
}
}
案例4：线程安全的资源池
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
public interface IResource : IDisposable
{
int Id { get; }
bool IsAvailable { get; }
void Use();
void Release();
}
public class MockResource : IResource
{
public int Id { get; }
public bool IsAvailable { get; private set; } = true;
public MockResource(int id)
{
Id = id;
Console.WriteLine($"资源 {Id} 创建");
}
public void Use()
{
if (!IsAvailable)
throw new InvalidOperationException($"资源 {Id} 不可用");
IsAvailable = false;
Console.WriteLine($"资源 {Id} 开始使用");
}
public void Release()
{
IsAvailable = true;
Console.WriteLine($"资源 {Id} 释放");
}
public void Dispose()
{
Console.WriteLine($"资源 {Id} 销毁");
}
}
public class ThreadSafeResourcePool
{
private readonly ConcurrentQueue
private readonly SemaphoreSlim semaphore;
private readonly Func
private readonly int maxSize;
private volatile bool disposed = false;
private int currentSize = 0;
public ThreadSafeResourcePool(Func
{
this.resourceFactory = resourceFactory ?? throw new ArgumentNullException(nameof(resourceFactory));
this.maxSize = maxSize;
this.semaphore = new SemaphoreSlim(maxSize, maxSize);
}
public async Task
{
if (disposed)
throw new ObjectDisposedException(nameof(ThreadSafeResourcePool
// 等待可用资源槽位
if (!await semaphore.WaitAsync(timeout))
{
throw new TimeoutException("获取资源超时");
}
try
{
// 尝试从池中获取现有资源
if (availableResources.TryDequeue(out T resource))
{
resource.Use();
Console.WriteLine($"从池中获取资源 {resource.Id}");
return resource;
}
// 如果池中没有资源，创建新资源
if (currentSize < maxSize)
{
resource = resourceFactory();
Interlocked.Increment(ref currentSize);
resource.Use();
Console.WriteLine($"创建新资源 {resource.Id}");
return resource;
}
throw new InvalidOperationException("无法创建更多资源");
}
catch
{
semaphore.Release(); // 出错时释放信号量
throw;
}
}
public T Acquire(TimeSpan timeout)
{
return AcquireAsync(timeout).GetAwaiter().GetResult();
}
public void Return(T resource)
{
if (resource == null)
throw new ArgumentNullException(nameof(resource));
if (disposed)
{
resource.Dispose();
return;
}
try
{
resource.Release();
availableResources.Enqueue(resource);
Console.WriteLine($"归还资源 {resource.Id} 到池中");
}
finally
{
semaphore.Release();
}
}
public int AvailableCount => availableResources.Count;
public int TotalCount => currentSize;
public void Dispose()
{
if (disposed) return;
disposed = true;
// 清理所有资源
while (availableResources.TryDequeue(out T resource))
{
resource.Dispose();
}
semaphore.Dispose();
Console.WriteLine("资源池已销毁");
}
}
class ResourcePoolExample
{
static async Task Main(string[] args)
{
Console.WriteLine("线程安全资源池测试");
// 创建资源池，最大5个资源
int resourceIdCounter = 0;
using (var resourcePool = new ThreadSafeResourcePool
() => new MockResource(Interlocked.Increment(ref resourceIdCounter)), 5))
{
// 启动多个任务竞争资源
Task[] tasks = new Task[10];
for (int i = 0; i < tasks.Length; i++)
{
int taskId = i;
tasks[i] = Task.Run(async () => await ResourceUser(resourcePool, taskId));
}
await Task.WhenAll(tasks);
Console.WriteLine($"测试完成 - 池中可用资源: {resourcePool.AvailableCount}, 总资源: {resourcePool.TotalCount}");
}
Console.WriteLine("资源池测试完成");
}
static async Task ResourceUser(ThreadSafeResourcePool
{
var random = new Random(userId);
for (int i = 0; i < 3; i++)
{
try
{
Console.WriteLine($"用户 {userId} 尝试获取资源...");
// 获取资源，设置5秒超时
var resource = await pool.AcquireAsync(TimeSpan.FromSeconds(5));
Console.WriteLine($"用户 {userId} 获得资源 {resource.Id}");
// 模拟使用资源
int useTime = random.Next(1000, 3000);
await Task.Delay(useTime);
Console.WriteLine($"用户 {userId} 使用资源 {resource.Id} 完成");
// 归还资源
pool.Return(resource);
}
catch (TimeoutException)
{
Console.WriteLine($"用户 {userId} 获取资源超时");
}
catch (Exception ex)
{
Console.WriteLine($"用户 {userId} 异常: {ex.Message}");
}
// 等待一段时间再次尝试
await Task.Delay(random.Next(500, 1500));
}
Console.WriteLine($"用户 {userId} 完成所有操作");
}
}
知识点总结
线程安全的核心原则：
避免数据竞争和竞态条件
保证操作的原子性
确保内存可见性
防止指令重排序
实现线程安全的策略：
不可变对象：最安全的方式
同步机制：lock、Monitor、读写锁等
原子操作：Interlocked类提供的原子操作
并发集合：ConcurrentDictionary、ConcurrentQueue等
线程本地存储：ThreadLocal
性能考虑：
选择合适的同步机制
避免过度同步
考虑读写比例选择读写锁
使用原子操作替代简单锁
设计模式：
单例模式的线程安全实现
生产者消费者模式
对象池模式
缓存模式
最佳实践：
优先使用不可变对象
合理使用并发集合
避免嵌套锁防止死锁
使用超时机制防止无限等待
定期清理过期资源
测试和调试：
进行充分的并发测试
使用压力测试验证性能
监控资源使用情况
记录和分析并发问题