同步的类型主要有两种: 通信和数据保护。
1. 阻塞锁
class LockClass
{
// 这个锁会保护 _value。
private readonly object _mutex = new object();
private int _value;
public void Increment()
{
lock (_mutex)
{
_value = _value + 1;
}
}
}
锁的使用,有四条重要的规则。
- 限制锁的作用范围。
- 文档中写清锁保护的内容。
- 锁范围内的代码尽量少。
- 在控制锁的时候绝不运行随意的代码
2. 异步锁
class SemaphoreSlimClass
{
// 这个锁保护 _value。
private readonly SemaphoreSlim _mutex = new SemaphoreSlim(1);
private int _value;
public async Task DelayAndIncrementAsync()
{
await _mutex.WaitAsync();
try
{
var oldValue = _value;
await Task.Delay(TimeSpan.FromSeconds(oldValue));
_value = oldValue + 1;
}
finally
{
_mutex.Release();
}
}
}
规则在这里也同样适用
- 限制锁的作用范围。
- 文档中写清锁保护的内容。
- 锁范围内的代码尽量少。
- 在控制锁的时候绝不运行随意的代码
3. 阻塞信号
class ManualResetEventSlimClass
{
private readonly ManualResetEventSlim _initialized = new ManualResetEventSlim();
private int _value;
public int WaitForInitialization()
{
_initialized.Wait();
return _value;
}
public void InitializeFromAnotherThread()
{
_value = 13;
_initialized.Set();
}
}
如果 ManualResetEventSlim
不能满足需求,还可考虑用 AutoResetEvent
、 CountdownEvent
或 Barrier
比喻:
-
ManualResetEventSlim
的整个工作方法有点像人群通过大门 -
AutoResetEvent
事件像一个旋转门,一次只允许一人通过。
4. 异步信号
class TaskCompletionSourceClass
{
private readonly TaskCompletionSource<object> _initialized = new TaskCompletionSource<object>();
private int _value1;
private int _value2;
public async Task<int> WaitForInitializationAsync()
{
await _initialized.Task;
return _value1 + _value2;
}
public void Initialize()
{
_value1 = 13;
_value2 = 17;
_initialized.TrySetResult(null);
}
}
信号是一种通用的通知机制
- 如果这个“信号”是一个用来在代码段之间发送数据的消息,那就考虑使用生产者/消费者队列
- 不要让通用的信号只是用来协调对共享数据的访问,那种情况下,可使用锁。
5. 限流
class MaxDegreeClass
{
//工作流限流
IPropagatorBlock<int, int> DataflowMultiplyBy2()
{
var options = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 10 };
return new TransformBlock<int, int>(data => data * 2, options);
}
// 使用 PLINQ
IEnumerable<int> ParallelMultiplyBy2(IEnumerable<int> values)
{
return values.AsParallel()
.WithDegreeOfParallelism(10)
.Select(item => item * 2);
}
// 使用 Parallel 类
void ParallelRotateMatrices(IEnumerable<Action<float>> matrices, float degrees)
{
var options = new ParallelOptions { MaxDegreeOfParallelism = 10 };
Parallel.ForEach(matrices, options, matrix => matrix.Invoke(degrees));
}
//并发性异步代码可以用 SemaphoreSlim 来限流:
async Task<string[]> DownloadUrlsAsync(
IEnumerable<string> urls)
{
var httpClient = new HttpClient();
var semaphore = new SemaphoreSlim(10);
var tasks = urls.Select(async url =>
{
await semaphore.WaitAsync();
try
{
return await httpClient.GetStringAsync(url);
}
finally
{
semaphore.Release();
}
}).ToArray();
return await Task.WhenAll(tasks);
}
}