使用 SemaphoreSlim 控制并发访问：从入门到实践

// 创建一个初始有 3 个许可证的信号量
// 也就是说：最多允许 3 个线程同时进入临界区
using var semaphore = new SemaphoreSlim(initialCount: 3);

// 模拟 10 个任务争抢 3 个名额
for (int i = 0; i < 10; i++)
{
    var index = i; // 闭包捕获需要局部变量
    Task.Run(async () =>
    {
        // Wait：申请一个许可证，没有就阻塞等待
        semaphore.Wait();
        try
        {
            Console.WriteLine($"任务 {index} 进入，开始工作");
            await Task.Delay(1000); // 模拟工作
            Console.WriteLine($"任务 {index} 完成");
        }
        finally
        {
            // Release：归还许可证，必须放在 finally 里
            // 否则一旦抛异常，许可证就再也还不回来，最终所有线程都会卡死
            semaphore.Release();
        }
    });
}

public class DatabaseAccess
{
    // 限制同时打开的数据库连接数
    private readonly SemaphoreSlim _connectionGate = new SemaphoreSlim(10);

    public async Task<Data> QueryAsync(string sql)
    {
        // WaitAsync：异步等待许可证，不阻塞线程
        await _connectionGate.WaitAsync();
        try
        {
            using var conn = new SqlConnection(_connectionString);
            await conn.OpenAsync();
            return await conn.QueryFirstOrDefaultAsync<Data>(sql);
        }
        finally
        {
            _connectionGate.Release();
        }
    }
}

public class Repository
{
    // 限制同时进行的数据库操作不超过 50
    // 这样即使有 500 个并发请求，也只有 50 个真正去拿连接
    private static readonly SemaphoreSlim DbGate = new SemaphoreSlim(50);

    public async Task<User> GetUserAsync(int id, CancellationToken ct)
    {
        await DbGate.WaitAsync(ct);
        try
        {
            // 真正占用连接的代码
            return await _db.Users.FindAsync(id);
        }
        finally
        {
            DbGate.Release();
        }
    }
}

public class ThirdPartyApiClient
{
    private readonly HttpClient _client;
    // 限制对外并发不超过 5，给第三方留余量
    private readonly SemaphoreSlim _rateGate = new SemaphoreSlim(5);

    public async Task<string> CallApiAsync(string url, CancellationToken ct)
    {
        await _rateGate.WaitAsync(ct);
        try
        {
            var resp = await _client.GetAsync(url, ct);
            resp.EnsureSuccessStatusCode();
            return await resp.Content.ReadAsStringAsync(ct);
        }
        finally
        {
            _rateGate.Release();
        }
    }
}

public class AsyncFileLogger
{
    // initialCount=1：同一时刻只允许一个写入
    private readonly SemaphoreSlim _writeLock = new SemaphoreSlim(1, 1);

    public async Task LogAsync(string message)
    {
        await _writeLock.WaitAsync();
        try
        {
            // 串行写入，避免内容交错
            await File.AppendAllTextAsync("app.log", $"{DateTime.Now:O} {message}\n");
        }
        finally
        {
            _writeLock.Release();
        }
    }
}

// 限制并发为 10，所有任务共享一个信号量
var sem = new SemaphoreSlim(10);
var tasks = urls.Select(async url =>
{
    await sem.WaitAsync();
    try { return await FetchAsync(url); }
    finally { sem.Release(); }
});
var results = await Task.WhenAll(tasks);

// 创建一个有界 Channel
var channel = Channel.CreateBounded<string>(100);
// 多个消费者从 Channel 读数据并处理
var consumers = Enumerable.Range(0, 10).Select(async _ =>
{
    await foreach (var url in channel.Reader.ReadAllAsync())
    {
        await FetchAsync(url);
    }
});
// 生产者把 URL 写入 Channel
foreach (var url in urls) await channel.Writer.WriteAsync(url);
channel.Writer.Complete();
await Task.WhenAll(consumers);

// 一个简单的可释放包装，配合 using 语句使用更优雅
public sealed class SemaphoreReleaser : IDisposable
{
    private readonly SemaphoreSlim _sem;
    public SemaphoreReleaser(SemaphoreSlim sem) { _sem = sem; }
    public void Dispose() => _sem.Release();
}

// 用法
public async Task DoWorkAsync()
{
    await _sem.WaitAsync();
    using var _ = new SemaphoreReleaser(_sem);
    // 业务逻辑
}

await _sem.WaitAsync(cancellationToken);