SemaphoreSlim节流

Question

我试图节流SemaphoreSlim (即允许使用负initialCount进行初始化)。有一种情况可能是，您正在访问一个API，并且可能会注意到由于服务器超载而导致的降级，因此您可能希望开始对请求进行节流，例如，10个并发请求。但是，您将对并发请求进行计数，并且知道此时有25个并发请求，但您无法启动一个SemaphoreSlim(-15，10)。

我尝试了一个允许这样做的实现，但我不能100%确定这是否是线程安全的，以及它是否可以被优化(例如，不使用锁)。

public class SemaphoreSlimThrottle : SemaphoreSlim
{
    private volatile int _throttleCount;
    private readonly object _lock = new object();

    public SemaphoreSlimThrottle(int initialCount)
        : base(initialCount)
    {
    }

    public SemaphoreSlimThrottle(int initialCount, int maxCount)
        : base(Math.Max(0, initialCount), maxCount)
    {
        _throttleCount = Math.Min(0, initialCount);
    }

    public new int CurrentCount => _throttleCount + base.CurrentCount;

    public new int Release()
    {
        if (_throttleCount < 0)
        {
            lock (_lock)
            {
                if (_throttleCount < 0)
                {
                    _throttleCount++;
                    return _throttleCount - 1;
                }
            }
        }
        return base.Release();
    }

    public new int Release(int releaseCount)
    {
        if (releaseCount < 1)
        {
            base.Release(releaseCount); // throws exception
        }

        if (releaseCount + _throttleCount <= 0)
        {
            lock (_lock)
            {
                if (releaseCount + _throttleCount <= 0)
                {
                    _throttleCount += releaseCount;
                    return _throttleCount - releaseCount;
                }
            }
        }

        if (_throttleCount < 0)
        {
            lock (_lock)
            {
                if (_throttleCount < 0)
                {
                    int output = CurrentCount;
                    base.Release(releaseCount + _throttleCount);
                    _throttleCount = 0;
                    return output;
                }
            }
        }

        return base.Release(releaseCount);
    }
}

我已经将它打包为一个NuGet包，其中包含了GitHub上可用的源代码。

Answer 1

首先是“新”关键字。如果将该类转换回SemaphoreSlim，则不使用具有新关键字的代码，而只执行基本SemaphoreSlim代码。

由于我们希望执行新代码，我建议不要继承SemaphoreSlim，而是包装它，然后将调用链接到包装好的SemaphoreSlim中。比如，我没有把所有的方法都放进去--这只是方法的一个例子。

public class SemaphoreSlimThrottle : IDisposable
{
    private volatile int _throttleCount;
    private readonly SemaphoreSlim _semaphore;
    private readonly object _lock = new object();
    private bool _turnOffThrottleCheck = false;

    public SemaphoreSlimThrottle(int initialCount, int maxCount)
    {
        if (initialCount < 0)
        {
            _semaphore = new SemaphoreSlim(0, maxCount);
            _turnOffThrottleCheck = true;
        }
        else
        {
            _semaphore = new SemaphoreSlim(initialCount, maxCount);
            _turnOffThrottleCheck = true;
        }
    }

    public WaitHandle AvailableWaitHandle => _semaphore.AvailableWaitHandle;

    public int CurrentCount => _throttleCount + _semaphore.CurrentCount;

    public bool Wait(TimeSpan timeout, CancellationToken cancellationToken) => _semaphore.Wait(timeout, cancellationToken);

    public int Release() => Release(1);

    public void Dispose()
    {
        _semaphore.Dispose();
    }
}

另一个可以预先设置的优化是，一旦“额外的”用完，就不能访问易失性字段。同时调用锁外的基座，使锁尽可能短。

警告--我没有测试这段代码--它可能有错误或错误--用作gist

。

    public int Release(int releaseCount)
    {
        // using bool property to avoid unnecessary volatile accesses in happy path
        if (releaseCount < 1 || _turnOffThrottleCheck)
        {
            return _semaphore.Release(releaseCount);
        }

        int remainingCount;
        var returnCount = 0;
        lock (_lock)
        {
            var throttleCount = _throttleCount;
            if (throttleCount == 0) // Different thread release all them just call into base
            {
                remainingCount = releaseCount;
            }
            else if (releaseCount + throttleCount < 0) // Releasing less than throttle just decrease
            {
                _throttleCount += releaseCount;
                remainingCount = 0;
                returnCount = throttleCount;
            }
            else // releasing all the throttles
            {
                _throttleCount = 0;
                _turnOffThrottleCheck = true;
                returnCount = throttleCount;
                remainingCount = releaseCount + throttleCount;
            }
        }

        // doing outside lock
        if (remainingCount > 0) // call into base if more locks to be released
        {
            return _semaphore.Release(releaseCount) + returnCount;
        }

        return returnCount + _semaphore.CurrentCount;

    }

这样，我们就不会在愉快的道路上访问_throttleCount。我们还释放锁，只需调用锁外的SemaphoreSlim。